Working version

app/controllers/__init__.py

@@ -1,15 +1,21 @@
 from litestar import Router
 
+from app.controllers.fiscal_payment_mapping import FiscalPaymentMappingController
+from app.controllers.machine import MachineController
 from app.domain.machine import Machine
 
-from . import machines
-
 __all__ = ["create_router"]
 
 
 def create_router() -> Router:
     return Router(
         path="/v1",
-        route_handlers=[machines.MachineController,],
+        route_handlers=[
-        signature_namespace={"Machine": Machine,},
+            MachineController,
+            FiscalPaymentMappingController,
+        ],
+        signature_namespace={
+            "Machine": Machine,
+            "FiscalPaymentMappingController": FiscalPaymentMappingController,
+        },
     )

app/controllers/fiscal_payment_mapping.py

@@ -0,0 +1,69 @@
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, Optional
+
+from litestar import Controller, get, post
+from litestar.contrib.repository.filters import SearchFilter
+from litestar.di import Provide
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from app.domain.fiscal_payment_mapping import (
+    FiscalPaymentMapping,
+    FiscalPaymentMappingReadDTO,
+    FiscalPaymentMappingWriteDTO,
+    Repository,
+    Service,
+)
+from app.lib.filters import ExactFilter
+from app.lib.responses import ObjectListResponse, ObjectResponse
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+DETAIL_ROUTE = "/{id:int}"
+
+
+def provides_service(db_session: AsyncSession) -> Service:
+    """Constructs repository and service objects for the request."""
+    return Service(Repository(session=db_session))
+
+
+class FiscalPaymentMappingController(Controller):
+    dto = FiscalPaymentMappingWriteDTO
+    return_dto = FiscalPaymentMappingReadDTO
+    path = "/fiscal-payment-mappings"
+    dependencies = {
+        "service": Provide(provides_service, sync_to_thread=False),
+    }
+    tags = ["FiscalPaymentMappings"]
+
+    @post()
+    async def create_fiscal_payment_mappings(
+        self, data: FiscalPaymentMapping, service: Service
+    ) -> FiscalPaymentMapping:
+        return await service.create(data)
+
+    @get()
+    async def get_fiscal_payment_mappings(
+        self, service: Service, payment_device_code: Optional[int] = None
+    ) -> ObjectListResponse[FiscalPaymentMapping]:
+        filters = []
+
+        if payment_device_code is not None:
+            filters.append(
+                ExactFilter[int](
+                    field_name="payment_device_code",
+                    value=payment_device_code,
+                ),
+            )
+
+        content = await service.list(*filters)
+        return ObjectListResponse(content=content)
+
+    @get(DETAIL_ROUTE)
+    async def get_fiscal_payment_mapping(
+        self, service: Service, id: int
+    ) -> ObjectResponse[FiscalPaymentMapping]:
+        content = await service.get(id)
+        return ObjectResponse(content=content)

app/controllers/machine.py

@@ -70,7 +70,7 @@ class MachineController(Controller):
             LimitOffset(limit=20, offset=0),
         ]
 
-        if search:
+        if search is not None:
             filters.append(
                 SearchFilter(
                     field_name="caption",

app/domain/fiscal_payment_mapping.py

@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+from enum import Enum
+from typing import Annotated, Optional
+
+import sqlalchemy
+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 ColumnElement
+from sqlalchemy.orm import Mapped, mapped_column
+
+from app.lib import service
+from app.lib.filters import ExactFilter
+
+
+class FiscalModuleEnum(str, Enum):
+    CROATIA = "CROATIA"
+    HUNGARY = "HUNGARY"
+    ITALY = "ITALY"
+    MONTENEGRO = "MONTENEGRO"
+    ROMANIA = "ROMANIA"
+    RUSSIA = "RUSSIA"
+    SERBIA = "SERBIA"
+
+
+class PaymentTypeEnum(str, Enum):
+    CA = "CA"
+    DA = "DA"
+    DB = "DB"
+    DC = "DC"
+    DD = "DD"
+    PA4 = "PA4"
+    NEG = "NEG"
+    PA3 = "PA3"
+    TA = "TA"
+    WLT = "WLT"
+
+
+class FiscalPaymentMapping(BigIntBase):
+    __tablename__ = "fiscal_payment_mapping"
+
+    fiscal_module: Mapped[FiscalModuleEnum] = mapped_column(
+        sqlalchemy.Enum(FiscalModuleEnum, name="fiscal_module_enum")
+    )
+    code: Mapped[str]
+    payment_type: Mapped[PaymentTypeEnum] = mapped_column(
+        sqlalchemy.Enum(PaymentTypeEnum, name="televend_payment_type")
+    )
+    operation_mode_code: Mapped[int]
+    payment_device_code: Mapped[Optional[int]]
+
+
+class Repository(SQLAlchemyAsyncRepository[FiscalPaymentMapping]):
+    model_type = FiscalPaymentMapping
+
+    def _apply_filters(
+        self, *filters: FilterTypes, apply_pagination: bool = True, statement: SelectT
+    ) -> SelectT:
+        standard_filters = []
+        for filter_ in filters:
+            if isinstance(filter_, ExactFilter):
+                field: ColumnElement = getattr(self.model_type, filter_.field_name)
+                statement = statement.where(field == filter_.value)
+            else:
+                standard_filters.append(filter_)
+
+        return super()._apply_filters(
+            *standard_filters, apply_pagination=apply_pagination, statement=statement
+        )
+
+
+class Service(service.Service[FiscalPaymentMapping]):
+    repository_type = Repository
+
+
+write_config = DTOConfig(exclude={"id"})
+FiscalPaymentMappingWriteDTO = SQLAlchemyDTO[
+    Annotated[FiscalPaymentMapping, write_config]
+]
+FiscalPaymentMappingReadDTO = SQLAlchemyDTO[FiscalPaymentMapping]

app/domain/machine.py

@@ -33,7 +33,7 @@ class Repository(SQLAlchemyAsyncRepository[Machine]):
         statement = super()._apply_filters(
             *filters, apply_pagination=apply_pagination, statement=statement
         )
-        statement = statement.where(Machine.alive == true())
+        statement = statement.where(self.model_type.alive == true())
         return statement
 
 

app/lib/filters.py

@@ -0,0 +1,10 @@
+from dataclasses import dataclass
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+@dataclass
+class ExactFilter(Generic[T]):
+    field_name: str
+    value: T