Implement new CRUD objects for Categories and Products

backend/app/alembic/versions/787343a5a78d_add_category_and_product_models.py

@@ -0,0 +1,47 @@
+"""Add Category and Product models
+
+Revision ID: 787343a5a78d
+Revises: 1a31ce608336
+Create Date: 2025-08-27 12:34:36.443129
+
+"""
+from alembic import op
+import sqlalchemy as sa
+import sqlmodel.sql.sqltypes
+
+
+# revision identifiers, used by Alembic.
+revision = '787343a5a78d'
+down_revision = '1a31ce608336'
+branch_labels = None
+depends_on = None
+
+
+def upgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.create_table('category',
+    sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+    sa.Column('description', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+    sa.Column('id', sa.Uuid(), nullable=False),
+    sa.Column('parent_category_id', sa.Uuid(), nullable=True),
+    sa.ForeignKeyConstraint(['parent_category_id'], ['category.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    op.create_table('product',
+    sa.Column('name', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=False),
+    sa.Column('description', sqlmodel.sql.sqltypes.AutoString(length=255), nullable=True),
+    sa.Column('price', sa.Float(), nullable=False),
+    sa.Column('discount_price', sa.Float(), nullable=True),
+    sa.Column('id', sa.Uuid(), nullable=False),
+    sa.Column('category_id', sa.Uuid(), nullable=False),
+    sa.ForeignKeyConstraint(['category_id'], ['category.id'], ),
+    sa.PrimaryKeyConstraint('id')
+    )
+    # ### end Alembic commands ###
+
+
+def downgrade():
+    # ### commands auto generated by Alembic - please adjust! ###
+    op.drop_table('product')
+    op.drop_table('category')
+    # ### end Alembic commands ###

backend/app/api/main.py

@@ -1,13 +1,15 @@
 from fastapi import APIRouter
 
-from app.api.routes import items, login, private, users, utils
+from app.api.routes import categories, items, login, private, products, users, utils
 from app.core.config import settings
 
 api_router = APIRouter()
 api_router.include_router(login.router)
 api_router.include_router(users.router)
 api_router.include_router(utils.router)
 api_router.include_router(items.router)
+api_router.include_router(categories.router)
+api_router.include_router(products.router)
 
 
 if settings.ENVIRONMENT == "local":

backend/app/api/routes/categories.py

@@ -0,0 +1,104 @@
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+from sqlmodel import func, select
+
+from app.api.deps import CurrentUser, SessionDep
+from app.models import (
+    Category,
+    CategoryCreate,
+    CategoryPublic,
+    CategoriesPublic,
+    CategoryUpdate,
+    Message,
+)
+
+router = APIRouter(prefix="/categories", tags=["categories"])
+
+
+@router.get("/", response_model=CategoriesPublic)
+def read_categories(
+    session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
+) -> Any:
+    """
+    Retrieve categories.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    count_statement = select(func.count()).select_from(Category)
+    count = session.exec(count_statement).one()
+    statement = select(Category).offset(skip).limit(limit)
+    categories = session.exec(statement).all()
+
+    return CategoriesPublic(data=categories, count=count)
+
+
+@router.get("/{id}", response_model=CategoryPublic)
+def read_category(session: SessionDep, current_user: CurrentUser, id: uuid.UUID) -> Any:
+    """
+    Get category by ID.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    category = session.get(Category, id)
+    if not category:
+        raise HTTPException(status_code=404, detail="Category not found")
+    return category
+
+
+@router.post("/", response_model=CategoryPublic)
+def create_category(
+    *, session: SessionDep, current_user: CurrentUser, category_in: CategoryCreate
+) -> Any:
+    """
+    Create new category.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    category = Category.model_validate(category_in)
+    session.add(category)
+    session.commit()
+    session.refresh(category)
+    return category
+
+
+@router.put("/{id}", response_model=CategoryPublic)
+def update_category(
+    *,
+    session: SessionDep,
+    current_user: CurrentUser,
+    id: uuid.UUID,
+    category_in: CategoryUpdate,
+) -> Any:
+    """
+    Update a category.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    category = session.get(Category, id)
+    if not category:
+        raise HTTPException(status_code=404, detail="Category not found")
+    update_dict = category_in.model_dump(exclude_unset=True)
+    category.sqlmodel_update(update_dict)
+    session.add(category)
+    session.commit()
+    session.refresh(category)
+    return category
+
+
+@router.delete("/{id}")
+def delete_category(
+    session: SessionDep, current_user: CurrentUser, id: uuid.UUID
+) -> Message:
+    """
+    Delete a category.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    category = session.get(Category, id)
+    if not category:
+        raise HTTPException(status_code=404, detail="Category not found")
+    session.delete(category)
+    session.commit()
+    return Message(message="Category deleted successfully")

backend/app/api/routes/products.py

@@ -0,0 +1,104 @@
+import uuid
+from typing import Any
+
+from fastapi import APIRouter, HTTPException
+from sqlmodel import func, select
+
+from app.api.deps import CurrentUser, SessionDep
+from app.models import (
+    Product,
+    ProductCreate,
+    ProductPublic,
+    ProductsPublic,
+    ProductUpdate,
+    Message,
+)
+
+router = APIRouter(prefix="/products", tags=["products"])
+
+
+@router.get("/", response_model=ProductsPublic)
+def read_products(
+    session: SessionDep, current_user: CurrentUser, skip: int = 0, limit: int = 100
+) -> Any:
+    """
+    Retrieve products.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    count_statement = select(func.count()).select_from(Product)
+    count = session.exec(count_statement).one()
+    statement = select(Product).offset(skip).limit(limit)
+    products = session.exec(statement).all()
+
+    return ProductsPublic(data=products, count=count)
+
+
+@router.get("/{id}", response_model=ProductPublic)
+def read_product(session: SessionDep, current_user: CurrentUser, id: uuid.UUID) -> Any:
+    """
+    Get product by ID.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    product = session.get(Product, id)
+    if not product:
+        raise HTTPException(status_code=404, detail="Product not found")
+    return product
+
+
+@router.post("/", response_model=ProductPublic)
+def create_product(
+    *, session: SessionDep, current_user: CurrentUser, product_in: ProductCreate
+) -> Any:
+    """
+    Create new product.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    product = Product.model_validate(product_in)
+    session.add(product)
+    session.commit()
+    session.refresh(product)
+    return product
+
+
+@router.put("/{id}", response_model=ProductPublic)
+def update_product(
+    *,
+    session: SessionDep,
+    current_user: CurrentUser,
+    id: uuid.UUID,
+    product_in: ProductUpdate,
+) -> Any:
+    """
+    Update a product.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    product = session.get(Product, id)
+    if not product:
+        raise HTTPException(status_code=404, detail="Product not found")
+    update_dict = product_in.model_dump(exclude_unset=True)
+    product.sqlmodel_update(update_dict)
+    session.add(product)
+    session.commit()
+    session.refresh(product)
+    return product
+
+
+@router.delete("/{id}")
+def delete_product(
+    session: SessionDep, current_user: CurrentUser, id: uuid.UUID
+) -> Message:
+    """
+    Delete a product.
+    """
+    if not current_user.is_superuser:
+        raise HTTPException(status_code=403, detail="Not enough permissions")
+    product = session.get(Product, id)
+    if not product:
+        raise HTTPException(status_code=404, detail="Product not found")
+    session.delete(product)
+    session.commit()
+    return Message(message="Product deleted successfully")

backend/app/models.py

@@ -92,6 +92,84 @@ class ItemsPublic(SQLModel):
     count: int
 
 
+# Category properties
+class CategoryBase(SQLModel):
+    name: str = Field(min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=255)
+
+
+# Properties to receive on category creation
+class CategoryCreate(CategoryBase):
+    pass
+
+
+# Properties to receive on category update
+class CategoryUpdate(CategoryBase):
+    name: str | None = Field(default=None, min_length=1, max_length=255)
+    parent_category_id: uuid.UUID | None = None
+
+
+# Database model
+class Category(CategoryBase, table=True):
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    parent_category_id: uuid.UUID | None = Field(
+        default=None, foreign_key="category.id"
+    )
+    parent_category: "Category" = Relationship(
+        back_populates="sub_categories", sa_relationship_kwargs=dict(remote_side="Category.id")
+    )
+    sub_categories: list["Category"] = Relationship(back_populates="parent_category")
+    products: list["Product"] = Relationship(back_populates="category")
+
+
+# Properties to return via API
+class CategoryPublic(CategoryBase):
+    id: uuid.UUID
+    parent_category_id: uuid.UUID | None
+
+
+class CategoriesPublic(SQLModel):
+    data: list[CategoryPublic]
+    count: int
+
+
+# Product properties
+class ProductBase(SQLModel):
+    name: str = Field(min_length=1, max_length=255)
+    description: str | None = Field(default=None, max_length=255)
+    price: float
+    discount_price: float | None = None
+
+
+# Properties to receive on product creation
+class ProductCreate(ProductBase):
+    category_id: uuid.UUID
+
+
+# Properties to receive on product update
+class ProductUpdate(ProductBase):
+    name: str | None = Field(default=None, min_length=1, max_length=255)
+    price: float | None = None
+
+
+# Database model
+class Product(ProductBase, table=True):
+    id: uuid.UUID = Field(default_factory=uuid.uuid4, primary_key=True)
+    category_id: uuid.UUID = Field(foreign_key="category.id")
+    category: Category = Relationship(back_populates="products")
+
+
+# Properties to return via API
+class ProductPublic(ProductBase):
+    id: uuid.UUID
+    category_id: uuid.UUID
+
+
+class ProductsPublic(SQLModel):
+    data: list[ProductPublic]
+    count: int
+
+
 # Generic message
 class Message(SQLModel):
     message: str