Companies

app/controllers/__init__.py

@@ -1,7 +1,10 @@
 from litestar import Router
 
+from app.controllers.company import CompanyController
 from app.controllers.fiscal_payment_mapping import FiscalPaymentMappingController
 from app.controllers.machine import MachineController
+from app.domain.company import Company
+from app.domain.fiscal_payment_mapping import FiscalPaymentMapping
 from app.domain.machine import Machine
 
 __all__ = ["create_router"]
@@ -11,11 +14,13 @@ def create_router() -> Router:
     return Router(
         path="/v1",
         route_handlers=[
+            CompanyController,
             MachineController,
             FiscalPaymentMappingController,
         ],
         signature_namespace={
+            "Company": Company,
             "Machine": Machine,
-            "FiscalPaymentMappingController": FiscalPaymentMappingController,
+            "FiscalPaymentMapping": FiscalPaymentMapping,
         },
     )

app/controllers/company.py

@@ -0,0 +1,64 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Optional
+
+from litestar import Controller, get
+from litestar.contrib.repository.filters import LimitOffset, SearchFilter
+from litestar.di import Provide
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.domain.company import (
+    Company,
+    CompanyReadDTO,
+    CompanyWriteDTO,
+    Repository,
+    Service,
+)
+from app.lib.responses import ObjectListResponse, ObjectResponse
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+DETAIL_ROUTE = "/{company_id:int}"
+
+
+def provides_service(db_session: AsyncSession) -> Service:
+    """Constructs repository and service objects for the request."""
+    return Service(Repository(session=db_session))
+
+
+class CompanyController(Controller):
+    dto = CompanyWriteDTO
+    return_dto = CompanyReadDTO
+    path = "/companies"
+    dependencies = {
+        "service": Provide(provides_service, sync_to_thread=False),
+    }
+    tags = ["Companies"]
+
+    @get()
+    async def get_companies(
+        self, service: Service, search: Optional[str] = None
+    ) -> ObjectListResponse[Company]:
+        filters = [
+            LimitOffset(limit=20, offset=0),
+        ]
+
+        if search is not None:
+            filters.append(
+                SearchFilter(
+                    field_name="caption",
+                    value=search,
+                ),
+            )
+
+        content = await service.list(*filters)
+        return ObjectListResponse(content=content)
+
+    @get(DETAIL_ROUTE)
+    async def get_company(
+        self, service: Service, company_id: int
+    ) -> ObjectResponse[Company]:
+        content = await service.get(company_id)
+        return ObjectResponse(content=content)

app/domain/company.py

@@ -0,0 +1,49 @@
+from __future__ import annotations
+
+from typing import Annotated
+
+from litestar.contrib.repository import FilterTypes
+from litestar.contrib.sqlalchemy.base import BigIntBase
+from litestar.contrib.sqlalchemy.dto import SQLAlchemyDTO
+from litestar.contrib.sqlalchemy.repository import SQLAlchemyAsyncRepository
+from litestar.contrib.sqlalchemy.repository.types import SelectT
+from litestar.dto import DTOConfig
+from sqlalchemy import true
+from sqlalchemy.orm import Mapped
+
+from app.lib import service
+
+
+class Company(BigIntBase):
+    __tablename__ = "vending_companies"
+
+    caption: Mapped[str]
+    address: Mapped[str]
+    city: Mapped[str]
+    phone: Mapped[str]
+    enabled: Mapped[str]
+    country_code: Mapped[str]
+    external_id: Mapped[str]
+    alive: Mapped[bool]
+
+
+class Repository(SQLAlchemyAsyncRepository[Company]):
+    model_type = Company
+
+    def _apply_filters(
+        self, *filters: FilterTypes, apply_pagination: bool = True, statement: SelectT
+    ) -> SelectT:
+        statement = super()._apply_filters(
+            *filters, apply_pagination=apply_pagination, statement=statement
+        )
+        statement = statement.where(self.model_type.alive == true())
+        return statement
+
+
+class Service(service.Service[Company]):
+    repository_type = Repository
+
+
+write_config = DTOConfig()
+CompanyWriteDTO = SQLAlchemyDTO[Annotated[Company, write_config]]
+CompanyReadDTO = SQLAlchemyDTO[Company]