102 lines
3.2 KiB
Python
102 lines
3.2 KiB
Python
from typing import List
|
|
|
|
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.ext.asyncio import AsyncSession
|
|
|
|
from app.dependencies.db import get_db
|
|
from app.schemas.category import (
|
|
CategoryCreate,
|
|
CategoryResponse,
|
|
CategoryUpdate,
|
|
CategoryWithProducts,
|
|
)
|
|
from app.services import category as category_service
|
|
|
|
router = APIRouter(prefix="/categories")
|
|
|
|
|
|
@router.get("", response_model=List[CategoryResponse], summary="Get all categories")
|
|
async def get_categories(db: AsyncSession = Depends(get_db)):
|
|
"""Get all categories."""
|
|
categories = await category_service.get_categories(db)
|
|
return categories
|
|
|
|
|
|
@router.get(
|
|
"/{category_id}", response_model=CategoryResponse, summary="Get a category by ID"
|
|
)
|
|
async def get_category(category_id: int, db: AsyncSession = Depends(get_db)):
|
|
"""Get a category by ID."""
|
|
db_category = await category_service.get_category(db, category_id)
|
|
if db_category is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Category with ID {category_id} not found",
|
|
)
|
|
return db_category
|
|
|
|
|
|
@router.get(
|
|
"/{category_id}/products",
|
|
response_model=CategoryWithProducts,
|
|
summary="Get a category with its products",
|
|
)
|
|
async def get_category_with_products(
|
|
category_id: int, db: AsyncSession = Depends(get_db)
|
|
):
|
|
"""Get a category with its related products."""
|
|
db_category = await category_service.get_category_with_products(db, category_id)
|
|
if db_category is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Category with ID {category_id} not found",
|
|
)
|
|
return db_category
|
|
|
|
|
|
@router.post(
|
|
"",
|
|
response_model=CategoryResponse,
|
|
status_code=status.HTTP_201_CREATED,
|
|
summary="Create a new category",
|
|
)
|
|
async def create_category(category: CategoryCreate, db: AsyncSession = Depends(get_db)):
|
|
"""Create a new category."""
|
|
return await category_service.create_category(db, category)
|
|
|
|
|
|
@router.put(
|
|
"/{category_id}", response_model=CategoryResponse, summary="Update a category"
|
|
)
|
|
async def update_category(
|
|
category_id: int,
|
|
category_update: CategoryUpdate,
|
|
db: AsyncSession = Depends(get_db),
|
|
):
|
|
"""Update an existing category."""
|
|
db_category = await category_service.get_category(db, category_id)
|
|
if db_category is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Category with ID {category_id} not found",
|
|
)
|
|
return await category_service.update_category(db, db_category, category_update)
|
|
|
|
|
|
@router.delete(
|
|
"/{category_id}",
|
|
status_code=status.HTTP_204_NO_CONTENT,
|
|
response_model=None,
|
|
summary="Delete a category",
|
|
)
|
|
async def delete_category(category_id: int, db: AsyncSession = Depends(get_db)):
|
|
"""Delete a category."""
|
|
db_category = await category_service.get_category(db, category_id)
|
|
if db_category is None:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_404_NOT_FOUND,
|
|
detail=f"Category with ID {category_id} not found",
|
|
)
|
|
await category_service.delete_category(db, db_category)
|
|
return None
|