elastic

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

@@ -26,6 +26,7 @@ services:
       POSTGRES_USER: user
       POSTGRES_PASSWORD: password
       POSTGRES_DB: fastapi_shop
+
     healthcheck:
       test: [ "CMD-SHELL", "pg_isready -U user -d fastapi_shop" ]
       interval: 10s

src/base_settings.py

@@ -17,7 +17,7 @@ class PostgresSettings(BaseModel):
     db: str = 'fastapi_shop'
     host: str = 'db'
     port: str = 5432
-    url: str = 'postgresql+asyncpg://:userpassword@db:5432/fastapi_shop'
+    url: str = 'postgresql+asyncpg://user:password@db:5432/fastapi_shop'
 
 
 class AuthorizationSettings(BaseModel):

src/catalogue/models/elasticsearch.py

@@ -5,6 +5,7 @@
 
 
 PRODUCT_INDEX = 'products_index'
+CATEGORY_INDEX = 'category_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 = PRODUCT_INDEX

src/catalogue/models/pydantic.py

@@ -5,3 +5,8 @@ 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
 
@@ -13,3 +13,13 @@ def __init__(self, session: AsyncSession):
 
 def get_product_repository(session: AsyncSession = Depends(get_session)) -> ProductRepository:
     return ProductRepository(session=session)
+
+
+
+class CategoryRepository(BaseSQLAlchemyRepository[Category]):
+    def __init__(self, session: AsyncSession):
+        super().__init__(model=Category, session=session)
+
+
+def get_category_repository(session: AsyncSession = Depends(get_session)) -> CategoryRepository:
+    return CategoryRepository(session=session)

src/catalogue/routes.py

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

src/catalogue/services.py

@@ -4,12 +4,12 @@
 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,
-    get_product_repository,
+    get_product_repository, get_category_repository, CategoryRepository,
 )
-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
@@ -45,3 +45,33 @@ async def update_search_index(self, uuid):
 
 def get_product_service(repo: ProductRepository = Depends(get_product_repository)) -> ProductService:
     return ProductService(repository=repo)
+
+
+
+
+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_category_service(repo: CategoryRepository = Depends(get_category_repository)) -> CategoryService:
+    return CategoryService(repository=repo)

src/catalogue/utils.py

@@ -7,12 +7,12 @@
 )
 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,
-    ProductIndex,
+    ProductIndex, CATEGORY_INDEX, CategoryIndex,
 )
-from src.catalogue.models.pydantic import ProductElasticResponse
+from src.catalogue.models.pydantic import ProductElasticResponse, CategoryElasticResponse
 from src.common.databases.elasticsearch import elastic_client
 
 
@@ -28,7 +28,7 @@ async def init_indices(self):
 
         products_index.document(ProductIndex)
 
-        if not await products_index.exists():
+        if not products_index.exists():
             await products_index.create()
 
     @staticmethod
@@ -79,3 +79,69 @@ 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):
+        category_index = Index(
+            name=CATEGORY_INDEX,
+            using=self.client,
+        )
+
+        category_index.document(CategoryIndex)
+
+        if not await category_index.exists():
+            await category_index.create()
+
+    @staticmethod
+    def build_category_search_query(keyword):
+        search = Search(
+            index='category_index',
+        ).query(
+            'multi_match',
+            query=keyword,
+            fields=['title', 'description'],
+        )
+        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,80 @@
+from typing import (
+    Annotated,
+    Union,
+)
+
+from fastapi import (
+    APIRouter,
+    BackgroundTasks,
+    Depends,
+    Response,
+    status,
+)
+
+from src.catalogue.models.database import Product, Category
+from src.catalogue.routes import (
+    CatalogueRoutesPrefixes,
+    ProductRoutesPrefixes,
+)
+from src.catalogue.services import get_product_service, get_category_service
+from src.common.enums import TaskStatus
+from src.general.schemas.task_status import TaskStatusModel
+
+
+router = APIRouter(prefix=CatalogueRoutesPrefixes.category)
+
+
+@router.get(
+    ProductRoutesPrefixes.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 products.
+
+    Returns:
+        Response with list of products.
+    """
+    return await category_service.list()
+
+
+
+@router.get(
+    ProductRoutesPrefixes.search,
+    status_code=status.HTTP_200_OK,
+)
+async def search(
+    keyword: str,
+    service: Annotated[get_category_service, Depends()],
+):
+    """
+    Search products.
+
+    Returns:
+        Response with products.
+    """
+    response = await service.search(keyword=keyword)
+
+    return response
+
+
+@router.post(
+    ProductRoutesPrefixes.update_index,
+    status_code=status.HTTP_200_OK,
+)
+async def update_elastic(
+    background_tasks: BackgroundTasks,
+    service: Annotated[get_category_service, Depends()],
+):
+    """
+    Update products index.
+
+    Returns:
+        None.
+    """
+    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

@@ -7,7 +7,7 @@
 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.views import product_router, category_router
 from src.common.databases.postgres import (
     engine,
     init_db,
@@ -44,6 +44,11 @@ def include_routes(application: FastAPI) -> None:
         prefix=BaseRoutesPrefixes.account,
         tags=['Account'],
     )
+    application.include_router(
+        router=category_router,
+        prefix=BaseRoutesPrefixes.catalogue,
+        tags=['Catalogue'],
+    )
 
 
 def get_application() -> FastAPI: