Desarrollo de una API REST para gestión de productos y valoraciones con FastAPI y MongoDB

{
    "name": "Producto 1",
    "description": "Descripción del producto 1",
    "price": 100,
    "categoryId": "60c72b2f9b1d8b3a4c8b4567"
}

{
    "name": "Categoría 1",
    "description": "Descripción de la categoría 1",
    "parentId": "60c72b2f9b1d8b3a4c8b4567"
}

{
    "productId": "60c72b2f9b1d8b3a4c8b4567",
    "userId": "60c72b2f9b1d8b3a4c8b4567",
    "username": "usuario123",
    "rating": 4.5,
    "comment": "Muy buen producto, la batería dura bastante."
}

from fastapi import FastAPI
from routes import test, categories # Importar el router de categorías

app = FastAPI(
    title="Products and Reviews API",
    description="API for managing products and reviews using FastAPI and MongoDB",
    version="1.0.0"
)

app.include_router(test.router, prefix="/api/test", tags=["test"])
app.include_router(categories.router, prefix="/api/categories", tags=["categories"])    # Añadir el router de categorías

db.categories.insertOne({
    name: "Categoría 1",
    description: "Descripción de la categoría 1",
    parentId: ObjectId("60c72b2f9b1d8b3a4c8b4567")
});

# Route to create a new category
@router.post("/", response_model=CategoryResponse, status_code=201)
async def create_category(category: Category):
    category_dict = category.dict()
    result = categories_collection.insert_one(category_dict)
    return {**category_dict, "id": str(result.inserted_id)}

db.categories.find();

# Define a function to build a category tree
def build_category_tree(categories, parent_id=None):
    branch = []
    for category in categories:
        if category["parentId"] == parent_id:
            children = build_category_tree(categories, category["id"])
            if children:
                category["children"] = children
            branch.append(category)
    return branch

# Route to get the category tree
@router.get("/tree", response_model=List[CategoryResponse])
async def get_category_tree():
    categories = list(categories_collection.find())
    for category in categories:
        category["id"] = str(category["_id"])
        category["parentId"] = str(category["parentId"]) if category["parentId"] else None
        del category["_id"]
    return build_category_tree(categories)

db.categories.find();

# Route to get all categories
@router.get("/", response_model=List[CategoryResponse])
async def get_categories():
    categories = list(categories_collection.find())
    for category in categories:
        category["id"] = str(category["_id"])
        del category["_id"]
    return categories

db.categories.findOne({ _id: ObjectId("60c72b2f9b1d8b3a4c8b4567") });

# Route to get a category by ID
@router.get("/{id}", response_model=CategoryResponse)
async def get_category(id: str):
    try:
        category = categories_collection.find_one({"_id": ObjectId(id)})
        if not category:
            raise HTTPException(status_code=404, detail="Category not found")
        category["id"] = str(category["_id"])
        del category["_id"]
        return category
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid category ID")

db.categories.updateOne(
    { _id: ObjectId("60c72b2f9b1d8b3a4c8b4567") },
    { $set: { name: "Nueva categoría", description: "Nueva descripción", parentId: ObjectId("60c72b2f9b1d8b3a4c8b4567") } }
);

# Route to update a category by ID
@router.put("/{id}", response_model=CategoryResponse)
async def update_category(id: str, category: Category):
    try:
        result = categories_collection.update_one(
            {"_id": ObjectId(id)},
            {"$set": category.dict()}
        )
        if result.matched_count == 0:
            raise HTTPException(status_code=404, detail="Category not found")
        updated_category = categories_collection.find_one({"_id": ObjectId(id)})
        updated_category["id"] = str(updated_category["_id"])
        del updated_category["_id"]
        return updated_category
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid category ID")

db.categories.deleteOne({ _id: ObjectId("60c72b2f9b1d8b3a4c8b4567") });

# Route to delete a category by ID
@router.delete("/{id}", status_code=200)
async def delete_category(id: str):
    try:
        result = categories_collection.delete_one({"_id": ObjectId(id)})
        if result.deleted_count == 0:
            raise HTTPException(status_code=404, detail="Category not found")
        return {"message": "Category deleted"}
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid category ID")

from fastapi import FastAPI
from routes import test, products, categories   # Importar el router de productos

app = FastAPI(
    title="Products and Reviews API",
    description="API for managing products and reviews using FastAPI and MongoDB",
    version="1.0.0"
)

app.include_router(test.router, prefix="/api/test", tags=["test"])
app.include_router(products.router, prefix="/api/products", tags=["products"]) # Añadir el router de productos
app.include_router(categories.router, prefix="/api/categories", tags=["categories"])

db.products.insertOne({
    name: "Producto 1",
    description: "Descripción del producto 1",
    price: 100,
    categoryId: ObjectId("60c72b2f9b1d8b3a4c8b4567")
});

# Route to create a new product
@router.post("/", response_model=ProductResponse, status_code=201)
async def create_product(product: Product):
    product_dict = product.dict()
    result = products_collection.insert_one(product_dict)
    return {**product_dict, "id": str(result.inserted_id)}

db.products.findOne({ _id: ObjectId("60c72b2f9b1d8b3a4c8b4567") });

# Route to get a product by ID
@router.get("/{id}", response_model=ProductResponse)
async def get_product(id: str):
    try:
        product = products_collection.find_one({"_id": ObjectId(id)})
        if not product:
            raise HTTPException(status_code=404, detail="Product not found")
        product["id"] = str(product["_id"])
        del product["_id"]
        return product
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid product ID")

db.products.find({ categoryId: ObjectId("60c72b2f9b1d8b3a4c8b4567") });

# Route to get all products
@router.get("/", response_model=List[ProductResponse])
async def get_products(categoryId: Optional[str] = None, name: Optional[str] = None):
    query = {}
    if categoryId:
        try:
            query["categoryId"] = categoryId
        except Exception:
            raise HTTPException(status_code=400, detail="Invalid category ID")
    if name:
        query["name"] = {"$regex": name, "$options": "i"}
    products = list(products_collection.find(query))
    for product in products:
        product["id"] = str(product["_id"])
        del product["_id"]
    return products

db.products.updateOne(
    { _id: ObjectId("60c72b2f9b1d8b3a4c8b4567") },
    { $set: {
        name: "Producto 2",
        description: "Descripción del producto 2",
        price: 150,
        categoryId: ObjectId("60c72b2f9b1d8b3a4c8b4568")
    }}
);

# Route to update a product by ID
@router.put("/{id}", response_model=ProductResponse)
async def update_product(id: str, product: Product):
    try:
        result = products_collection.update_one(
            {"_id": ObjectId(id)},
            {"$set": product.dict()}
        )
        if result.matched_count == 0:
            raise HTTPException(status_code=404, detail="Product not found")
        updated_product = products_collection.find_one({"_id": ObjectId(id)})
        updated_product["id"] = str(updated_product["_id"])
        del updated_product["_id"]
        return updated_product
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid product ID")

db.products.deleteOne({ _id: ObjectId("60c72b2f9b1d8b3a4c8b4567") });

# Route to delete a product by ID
@router.delete("/{id}", status_code=200)
async def delete_product(id: str):
    try:
        result = products_collection.delete_one({"_id": ObjectId(id)})
        if result.deleted_count == 0:
            raise HTTPException(status_code=404, detail="Product not found")
        reviews_collection.delete_many({"productId": ObjectId(id)})
        return {"message": "Product and associated reviews deleted"}
    except Exception:
        raise HTTPException(status_code=400, detail="Invalid product ID")

from fastapi import FastAPI
from routes import test, products, categories, reviews  # Importar el router de comentarios

app = FastAPI(
    title="Products and Reviews API",
    description="API for managing products and reviews using FastAPI and MongoDB",
    version="1.0.0"
)

app.include_router(test.router, prefix="/api/test", tags=["test"])
app.include_router(products.router, prefix="/api/products", tags=["products"])
app.include_router(categories.router, prefix="/api/categories", tags=["categories"])
app.include_router(reviews.router, prefix="/api/reviews", tags=["reviews"])  # Añadir el router de comentarios

db.reviews.insertOne({
    productId: ObjectId("60c72b2f9b1d8b3a4c8b4567"),
    userId: ObjectId("60c72b2f9b1d8b3a4c8b4568"),
    username: "usuario123",
    rating: 4.5,
    comment: "Muy buen producto, la batería dura bastante.",
    createdAt: new Date()
});

# Route to create a new review
@router.post("/", response_model=ReviewResponse, status_code=201)
async def create_review(review: Review):
    try:
        review_dict = review.dict()
        review_dict["productId"] = ObjectId(review.productId)
        review_dict["userId"] = ObjectId(review.userId)
        review_dict["createdAt"] = datetime.now(pytz.UTC)

        result = reviews_collection.insert_one(review_dict)
        review_dict["id"] = str(result.inserted_id)
        review_dict["productId"] = str(review_dict["productId"])
        review_dict["userId"] = str(review_dict["userId"])
        review_dict["createdAt"] = review_dict["createdAt"].strftime("%Y-%m-%dT%H:%M:%SZ")

        return review_dict
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

db.reviews.find({ productId: ObjectId("60c72b2f9b1d8b3a4c8b4567") });

# Route to get all reviews or filter by productId or userId
@router.get("/", response_model=List[ReviewResponse])
async def get_reviews(productId: Optional[str] = None, userId: Optional[str] = None):
    try:
        query = {}
        if productId:
            query["productId"] = ObjectId(productId)
        if userId:
            query["userId"] = ObjectId(userId)

        reviews = list(reviews_collection.find(query))
        for review in reviews:
            review["id"] = str(review.pop("_id"))
            review["productId"] = str(review["productId"])
            review["userId"] = str(review["userId"])
            review["createdAt"] = review["createdAt"].strftime("%Y-%m-%dT%H:%M:%SZ")
            if "updatedAt" in review:
                review["updatedAt"] = review["updatedAt"].strftime("%Y-%m-%dT%H:%M:%SZ")
        return reviews
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

db.reviews.updateOne(
    { _id: ObjectId("60c72b2f9b1d8b3a4c8b4567"), userId: ObjectId("60c72b2f9b1d8b3a4c8b4568") },
    { $set: {
        rating: 4.0,
        comment: "Muy buen producto, la batería dura bastante.",
        updatedAt: new Date()
    }}
);

# Route to update a review by ID
@router.put("/{id}", response_model=ReviewResponse)
async def update_review(id: str, review: Review):
    try:
        existing_review = reviews_collection.find_one({
            "_id": ObjectId(id),
            "userId": ObjectId(review.userId)
        })

        if not existing_review:
            raise HTTPException(status_code=403, detail="You can only update your own reviews")

        update_data = {
            "rating": review.rating,
            "comment": review.comment,
            "updatedAt": datetime.now(pytz.UTC)
        }

        result = reviews_collection.update_one(
            {"_id": ObjectId(id)},
            {"$set": update_data}
        )

        if result.matched_count == 0:
            raise HTTPException(status_code=404, detail="Review not found")

        updated_review = reviews_collection.find_one({"_id": ObjectId(id)})
        updated_review["id"] = str(updated_review.pop("_id"))
        updated_review["productId"] = str(updated_review["productId"])
        updated_review["userId"] = str(updated_review["userId"])
        updated_review["createdAt"] = updated_review["createdAt"].strftime("%Y-%m-%dT%H:%M:%SZ")
        updated_review["updatedAt"] = updated_review["updatedAt"].strftime("%Y-%m-%dT%H:%M:%SZ")

        return updated_review
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))

db.reviews.deleteOne({ _id: ObjectId("60c72b2f9b1d8b3a4c8b4567") });

# Route to delete a review by ID
@router.delete("/{id}")
async def delete_review(id: str, request_body: dict):
    try:
        userId = request_body.get("userId")
        if not userId:
            raise HTTPException(status_code=400, detail="userId is required in request body")

        existing_review = reviews_collection.find_one({
            "_id": ObjectId(id),
            "userId": ObjectId(userId)
        })

        if not existing_review:
            raise HTTPException(status_code=403, detail="You can only delete your own reviews")

        result = reviews_collection.delete_one({"_id": ObjectId(id)})

        if result.deleted_count == 0:
            raise HTTPException(status_code=404, detail="Review not found")

        return {"message": "Review deleted"}
    except Exception as e:
        raise HTTPException(status_code=400, detail=str(e))