83 lines
2.4 KiB
Python
83 lines
2.4 KiB
Python
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" # type: ignore[assignment]
|
|
|
|
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]
|