hw lesson 26

Dockerfile

@@ -2,12 +2,14 @@ FROM python:3.11
 
 WORKDIR /app
 
-COPY pyproject.toml pyproject.toml
+COPY pyproject.toml poetry.lock ./
 
 RUN pip install poetry
 RUN poetry config virtualenvs.create false
-RUN poetry install --no-dev
+RUN poetry config virtualenvs.in-project false
+
+RUN poetry install --only=main --no-interaction --no-ansi --no-root
 
 COPY . /app
 
-CMD ["sh", "ops/start-api.sh"]
+CMD ["sh", "ops/start-api.sh"]

docker-compose.yml

@@ -20,6 +20,8 @@ services:
 
   db:
     image: postgres:13
+    ports:
+      - "5432:5432"
     volumes:
       - postgres_data:/var/lib/postgresql/data
     environment:

src/catalogue/models/elasticsearch.py

@@ -5,6 +5,7 @@
 
 
 PRODUCT_INDEX = 'products_index'
+CATEGORY_INDEX = 'categories_index'
 
 
 class ProductIndex(Document):
@@ -14,3 +15,11 @@ class ProductIndex(Document):
 
     class Index:
         name = PRODUCT_INDEX
+
+
+class CategoryIndex(Document):
+    title = Text()
+    description = Text()
+
+    class Index:
+        name = CATEGORY_INDEX

src/catalogue/models/pydantic.py

@@ -5,3 +5,9 @@ class ProductElasticResponse(BaseModel):
     product_id: int
     title: str
     score: float
+
+
+class CategoryElasticResponse(BaseModel):
+    category_id: int
+    title: str
+    score: float

src/catalogue/repository.py

@@ -1,7 +1,7 @@
 from fastapi import Depends
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from src.catalogue.models.database import Product
+from src.catalogue.models.database import Product, Category
 from src.common.databases.postgres import get_session
 from src.common.repository.sqlalchemy import BaseSQLAlchemyRepository
 
@@ -11,5 +11,14 @@ def __init__(self, session: AsyncSession):
         super().__init__(model=Product, session=session)
 
 
+class CategoryRepository(BaseSQLAlchemyRepository[Category]):
+    def __init__(self, session: AsyncSession):
+        super().__init__(model=Category, session=session)
+
+
 def get_product_repository(session: AsyncSession = Depends(get_session)) -> ProductRepository:
     return ProductRepository(session=session)
+
+
+def get_category_repository(session: AsyncSession = Depends(get_session)) -> CategoryRepository:
+    return CategoryRepository(session=session)

src/catalogue/routes.py

@@ -3,8 +3,14 @@
 
 class CatalogueRoutesPrefixes:
     product: str = '/product'
+    category: str = '/category'
 
 
 class ProductRoutesPrefixes(BaseCrudPrefixes):
     search: str = '/search'
     update_index: str = '/update-index'
+
+
+class CategoryRoutesPrefixes(BaseCrudPrefixes):
+    search: str = '/search'
+    update_index: str = '/update-index'

src/catalogue/services.py

@@ -4,12 +4,14 @@
 from fastapi import Depends
 
 from src.base_settings import base_settings
-from src.catalogue.models.database import Product
+from src.catalogue.models.database import Product, Category
 from src.catalogue.repository import (
     ProductRepository,
+    CategoryRepository,
     get_product_repository,
+    get_category_repository,
 )
-from src.catalogue.utils import ProductElasticManager
+from src.catalogue.utils import ProductElasticManager, CategoryElasticManager
 from src.common.enums import TaskStatus
 from src.common.service import BaseService
 from src.general.schemas.task_status import TaskStatusModel
@@ -43,5 +45,33 @@ async def update_search_index(self, uuid):
         ).save_to_redis()
 
 
+class CategoryService(BaseService[Category]):
+    def __init__(self, repository: CategoryRepository):
+        super().__init__(repository)
+
+    @staticmethod
+    async def search(keyword: str):
+        result = await CategoryElasticManager().search_category(keyword=keyword)
+        return result
+
+    async def update_category_index(self, uuid):
+        categories = await self.list()
+
+        try:
+            await CategoryElasticManager().update_index(categories=categories)
+        except ConnectionError as exc:
+            await TaskStatusModel(uuid=uuid, status=TaskStatus.ERROR, details=str(exc)).save_to_redis()
+
+        await TaskStatusModel(
+            uuid=uuid,
+            status=TaskStatus.DONE,
+            done_at=datetime.utcnow().strftime(base_settings.date_time_format),
+        ).save_to_redis()
+
+
 def get_product_service(repo: ProductRepository = Depends(get_product_repository)) -> ProductService:
     return ProductService(repository=repo)
+
+
+def get_category_service(repo: CategoryRepository = Depends(get_category_repository)) -> CategoryService:
+    return CategoryService(repository=repo)

src/catalogue/utils.py

@@ -7,12 +7,14 @@
 )
 from fastapi import Depends
 
-from src.catalogue.models.database import Product
+from src.catalogue.models.database import Product, Category
 from src.catalogue.models.elasticsearch import (
     PRODUCT_INDEX,
+    CATEGORY_INDEX,
     ProductIndex,
+    CategoryIndex,
 )
-from src.catalogue.models.pydantic import ProductElasticResponse
+from src.catalogue.models.pydantic import ProductElasticResponse, CategoryElasticResponse
 from src.common.databases.elasticsearch import elastic_client
 
 
@@ -39,6 +41,7 @@ def build_product_search_query(keyword):
             'multi_match',
             query=keyword,
             fields=['title', 'description', 'short_description'],
+            fuzziness='AUTO',
         )
         return search.to_dict()
 
@@ -79,3 +82,68 @@ async def update_index(self, products: list[Product]) -> None:
 
         if bulk_data:
             await self.client.bulk(body=bulk_data)
+
+
+class CategoryElasticManager:
+    def __init__(self, client: Annotated[AsyncElasticsearch, Depends(elastic_client)] = elastic_client):
+        self.client = client
+
+    async def init_indices(self):
+        categories_index = Index(
+            name=CATEGORY_INDEX,
+            using=self.client,
+        )
+
+        categories_index.document(CategoryIndex)
+
+        if not await categories_index.exists():
+            await categories_index.create()
+
+    @staticmethod
+    def build_category_search_query(keyword):
+        search = Search(
+            index='categories_index',
+        ).query(
+            'multi_match',
+            query=keyword,
+            fields=['title', 'description'],
+            fuzziness='AUTO',
+        )
+        return search.to_dict()
+
+    async def search_category(self, keyword):
+        query = self.build_category_search_query(keyword)
+        response = await self.client.search(body=query)
+        await self.client.close()
+
+        hits = response.get('hits', {}).get('hits', [])
+        sorted_hits = sorted(hits, key=lambda x: x.get('_score', 0), reverse=True)
+
+        sorted_response = [
+            CategoryElasticResponse(
+                category_id=hit.get('_id', ''),
+                title=hit.get('_source', {}).get('title', ''),
+                score=hit.get('_score', {}),
+            )
+            for hit in sorted_hits
+        ]
+
+        return sorted_response
+
+    async def update_index(self, categories: list[Category]) -> None:
+        bulk_data = []
+        for category in categories:
+            action = {'index': {'_index': CATEGORY_INDEX, '_id': category.id}}
+            data = {
+                'title': category.title,
+                'description': category.description,
+            }
+            bulk_data.append(action)
+            bulk_data.append(data)
+
+            if len(bulk_data) >= 100:
+                await self.client.bulk(body=bulk_data)
+                bulk_data = []
+
+        if bulk_data:
+            await self.client.bulk(body=bulk_data)

src/catalogue/views/__init__.py

@@ -1 +1,2 @@
 from .product import router as product_router
+from .category import router as category_router

src/catalogue/views/category.py

@@ -0,0 +1,110 @@
+from typing import (
+    Annotated,
+    Union,
+)
+
+from fastapi import (
+    APIRouter,
+    BackgroundTasks,
+    Depends,
+    Response,
+    status,
+)
+
+from src.catalogue.models.database import Category
+from src.catalogue.routes import (
+    CatalogueRoutesPrefixes,
+    CategoryRoutesPrefixes,
+)
+from src.catalogue.services import get_category_service
+from src.common.enums import TaskStatus
+from src.common.exceptions.base import ObjectDoesNotExistException
+from src.common.schemas.common import ErrorResponse
+from src.general.schemas.task_status import TaskStatusModel
+
+
+router = APIRouter(prefix=CatalogueRoutesPrefixes.category)
+
+
+@router.get(
+    CategoryRoutesPrefixes.root,
+    status_code=status.HTTP_200_OK,
+    response_model=list[Category],
+)
+async def category_list(category_service: Annotated[get_category_service, Depends()]) -> list[Category]:
+    """
+    Get list of categories.
+
+    Returns:
+        Response with list of categories.
+    """
+    return await category_service.list()
+
+
+@router.get(
+    CategoryRoutesPrefixes.detail,
+    responses={
+        status.HTTP_200_OK: {'model': Category},
+        status.HTTP_404_NOT_FOUND: {'model': ErrorResponse},
+    },
+    status_code=status.HTTP_200_OK,
+    response_model=Union[Category, ErrorResponse],
+)
+async def category_detail(
+    response: Response,
+    pk: int,
+    service: Annotated[get_category_service, Depends()],
+) -> Union[Response, ErrorResponse]:
+    """
+    Retrieve category.
+
+    Returns:
+        Response with category details.
+    """
+    try:
+        response = await service.detail(pk=pk)
+    except ObjectDoesNotExistException as exc:
+        response.status_code = status.HTTP_404_NOT_FOUND
+        return ErrorResponse(message=exc.message)
+
+    return response
+
+
+@router.get(
+    CategoryRoutesPrefixes.search,
+    status_code=status.HTTP_200_OK,
+)
+async def search(
+    keyword: str,
+    service: Annotated[get_category_service, Depends()],
+):
+    """
+    Search categories.
+
+    Returns:
+        Response with categories.
+    """
+    response = await service.search(keyword=keyword)
+
+    return response
+
+
+@router.post(
+    CategoryRoutesPrefixes.update_index,
+    status_code=status.HTTP_200_OK,
+)
+async def update_elastic(
+    background_tasks: BackgroundTasks,
+    service: Annotated[get_category_service, Depends()],
+):
+    """
+    Update categories index.
+
+    Returns:
+        Task status.
+    """
+    status_model = await TaskStatusModel(status=TaskStatus.IN_PROGRESS).save_to_redis()
+
+    background_tasks.add_task(service.update_category_index, status_model.uuid)
+
+    return await TaskStatusModel().get_from_redis(uuid=status_model.uuid)

src/main.py

@@ -6,8 +6,8 @@
 from src.admin import register_admin_views
 from src.authentication.views import router as auth_router
 from src.base_settings import base_settings
-from src.catalogue.utils import ProductElasticManager
-from src.catalogue.views import product_router
+from src.catalogue.utils import ProductElasticManager, CategoryElasticManager
+from src.catalogue.views import product_router, category_router
 from src.common.databases.postgres import (
     engine,
     init_db,
@@ -21,6 +21,7 @@
 async def lifespan(application: FastAPI):  # noqa: ARG001
     await init_db()
     await ProductElasticManager().init_indices()
+    await CategoryElasticManager().init_indices()
 
     yield
 
@@ -39,6 +40,11 @@ def include_routes(application: FastAPI) -> None:
         prefix=BaseRoutesPrefixes.catalogue,
         tags=['Catalogue'],
     )
+    application.include_router(
+        router=category_router,
+        prefix=BaseRoutesPrefixes.catalogue,
+        tags=['Catalogue'],
+    )
     application.include_router(
         router=user_router,
         prefix=BaseRoutesPrefixes.account,