hw lesson 21 Category model, repository, service, api routes

Dockerfile

@@ -6,7 +6,7 @@ COPY pyproject.toml pyproject.toml
 
 RUN pip install poetry
 RUN poetry config virtualenvs.create false
-RUN poetry install --no-dev
+RUN poetry install --no-root
 
 COPY . /app
 

docker-compose.yml

@@ -1,5 +1,3 @@
-version: '3.8'
-
 services:
   web:
     build: .

src/catalogue/models/pydantic.py

@@ -15,3 +15,15 @@ class ProductModel(BaseModel):
 
     class Config:
         from_attributes = True
+
+
+class CategoryModel(BaseModel):
+    id: Optional[int]
+    title: str
+    description: Optional[str]
+    image: Optional[str]
+    is_active: bool
+    parent_id: Optional[int]
+
+    class Config:
+        from_attributes = True

src/catalogue/repository.py

@@ -1,8 +1,8 @@
 from fastapi import Depends
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.catalogue.models.pydantic import ProductModel
-from src.catalogue.models.sqlalchemy import Product
+from src.catalogue.models.pydantic import ProductModel, CategoryModel
+from src.catalogue.models.sqlalchemy import Product, Category
 from src.common.databases.postgres import (
     get_session,
 )
@@ -16,3 +16,12 @@ def __init__(self, session: AsyncSession):
 
 def get_product_repository(session: AsyncSession = Depends(get_session)) -> ProductRepository:
     return ProductRepository(session=session)
+
+
+class CategoryRepository(BaseSQLAlchemyRepository[Category, CategoryModel]):
+    def __init__(self, session: AsyncSession):
+        super().__init__(model=Category, pydantic_model=CategoryModel, session=session)
+
+
+def get_category_repository(session: AsyncSession = Depends(get_session)) -> CategoryRepository:
+    return CategoryRepository(session=session)

src/catalogue/routes.py

@@ -3,7 +3,12 @@
 
 class CatalogueRoutesPrefixes:
     product: str = '/product'
+    category: str = '/category'
 
 
 class ProductRoutesPrefixes(BaseCrudPrefixes):
     ...
+
+
+class CategoryRoutesPrefixes(BaseCrudPrefixes):
+    ...

src/catalogue/services.py

@@ -1,9 +1,12 @@
 from fastapi import Depends
+from sqlalchemy import update as sqlalchemy_update
 
-from src.catalogue.models.pydantic import ProductModel
+from src.catalogue.models.pydantic import ProductModel, CategoryModel
 from src.catalogue.repository import (
     ProductRepository,
     get_product_repository,
+    CategoryRepository,
+    get_category_repository,
 )
 from src.common.service import BaseService
 
@@ -15,3 +18,29 @@ def __init__(self, repository: ProductRepository):
 
 def get_product_service(repo: ProductRepository = Depends(get_product_repository)) -> ProductService:
     return ProductService(repository=repo)
+
+
+class CategoryService(BaseService[CategoryModel]):
+    def __init__(self, repository: CategoryRepository):
+        super().__init__(repository)
+
+    async def list_active(self) -> list[CategoryModel]:
+        return await self.repository.filter(is_active=True)
+
+    async def list_subcategories(self, parent_id: int) -> list[CategoryModel]:
+        return await self.repository.filter(parent_id=parent_id)
+
+    async def update(self, pk: int, update_data: CategoryModel) -> CategoryModel:
+        if update_data.parent_id is not None and update_data.parent_id == pk:
+            raise ValueError("Category cannot be parent of itself.")
+
+        values = update_data.model_dump(exclude={"id"})
+        await self.repository.session.execute(
+            sqlalchemy_update(self.repository.model).where(self.repository.model.id == pk).values(**values)
+        )
+        await self.repository.session.commit()
+        return await self.repository.get(pk=pk)
+
+
+def get_category_service(repo: CategoryRepository = Depends(get_category_repository)) -> CategoryService:
+    return CategoryService(repository=repo)

src/catalogue/views.py

@@ -10,19 +10,22 @@
     status,
 )
 
-from src.catalogue.models.pydantic import ProductModel
+from src.catalogue.models.pydantic import ProductModel, CategoryModel
 from src.catalogue.routes import (
     CatalogueRoutesPrefixes,
     ProductRoutesPrefixes,
+    CategoryRoutesPrefixes,
 )
 from src.catalogue.services import (
     get_product_service,
+    get_category_service,
 )
 from src.common.exceptions.base import ObjectDoesNotExistException
 from src.common.schemas.common import ErrorResponse
 
 
 product_router = APIRouter(prefix=CatalogueRoutesPrefixes.product)
+category_router = APIRouter(prefix=CatalogueRoutesPrefixes.category)
 
 
 @product_router.get(
@@ -69,3 +72,41 @@ async def product_detail(
         return ErrorResponse(message=exc.message)
 
     return response
+
+
+@category_router.get(
+    CategoryRoutesPrefixes.root,
+    status_code=status.HTTP_200_OK,
+    response_model=list[CategoryModel],
+)
+async def category_list(service=Depends(get_category_service)) -> list[CategoryModel]:
+    """
+    Get list of categories.
+    """
+    return await service.list()
+
+
+@category_router.get(
+    CategoryRoutesPrefixes.detail,
+    responses={
+        status.HTTP_200_OK: {'model': CategoryModel},
+        status.HTTP_404_NOT_FOUND: {'model': ErrorResponse},
+    },
+    status_code=status.HTTP_200_OK,
+    response_model=Union[CategoryModel, ErrorResponse],
+)
+async def category_detail(
+    response: Response,
+    pk: int,
+    service: Annotated[get_category_service, Depends()],
+) -> Union[Response, ErrorResponse]:
+    """
+    Retrieve category.
+    """
+    try:
+        result = await service.detail(pk=pk)
+    except ObjectDoesNotExistException as exc:
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return ErrorResponse(message=exc.message)
+
+    return result

src/common/service.py

@@ -26,7 +26,7 @@ async def create(self, instance_data: PType) -> PType:
         return await self.repository.create(instance_data)
 
     async def update(self, pk: int, update_data: PType) -> PType:
-        return await self.repository.update(id, update_data)
+        return await self.repository.update(pk, update_data)
 
     async def delete(self, pk: int):
         await self.repository.delete(pk=pk)

src/main.py

@@ -3,7 +3,7 @@
 
 from src.admin import register_admin_views
 from src.base_settings import base_settings
-from src.catalogue.views import product_router
+from src.catalogue.views import product_router, category_router
 from src.common.databases.postgres import postgres
 from src.general.views import router as status_router
 from src.routes import BaseRoutesPrefixes
@@ -18,6 +18,11 @@ def include_routes(application: FastAPI) -> None:
         prefix=BaseRoutesPrefixes.catalogue,
         tags=['Catalogue'],
     )
+    application.include_router(
+        router=category_router,
+        prefix=BaseRoutesPrefixes.catalogue,
+        tags=['Catalogue'],
+    )
 
 
 def get_application() -> FastAPI: