Working version

app/controllers/machines.py

@@ -1,8 +1,10 @@
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Optional, cast
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Optional, cast, TypeVar, Type, Generic
 
 from litestar import Controller, delete, get, post, put
+from litestar.contrib.repository.filters import LimitOffset, SearchFilter, FilterTypes
 from litestar.di import Provide
 from litestar.pagination import (
     AbstractAsyncOffsetPaginator,
@@ -12,16 +14,19 @@ from litestar.status_codes import HTTP_200_OK
 from sqlalchemy import ScalarResult, func, select
 from sqlalchemy.ext.asyncio import AsyncSession
 
-from app.domain.machine import MachineReadDTO, MachineWriteDTO, Repository, Service
+from app.domain.machine import (
-from app.domain.machine import Machine
+    Machine,
-
+    MachineReadDTO,
-from litestar.contrib.repository.filters import SearchFilter, LimitOffset
+    MachineWriteDTO,
+    Repository,
+    Service,
+)
+from app.lib.responses import ObjectListResponse, ObjectResponse
 
 if TYPE_CHECKING:
     from sqlalchemy.ext.asyncio import AsyncSession
 
 
-
 DETAIL_ROUTE = "/{machine_id:int}"
 
 
@@ -59,20 +64,10 @@ class MachineController(Controller):
 
     @get()
     async def get_machines(
-        self, service: Service,
+        self, service: Service, search: Optional[str] = None
-        search: Optional[str] = None,
+    ) -> ObjectListResponse[Machine]:
-    ) -> list[Machine]:
-        """Get a list of authors."""
-
-        print("#" * 100)
-        print(search)
-        print("#" * 100)
-
         filters = [
-            LimitOffset(
+            LimitOffset(limit=20, offset=0),
-                limit=20,
-                offset=0
-            ),
         ]
 
         if search:
@@ -83,22 +78,18 @@ class MachineController(Controller):
                 ),
             )
 
-        return await service.list(*filters)
+        content = await service.list(*filters)
-
+        return ObjectListResponse(content=content)
-    # @get()
-    # async def get_machines(
-    #     self, service: Service, filters: list[FilterTypes]
-    # ) -> list[Machine]:
-    #     """Get a list of authors."""
-    #     return await service.list(*filters)
 
     # @post()
     # async def create_author(self, data: Machine, service: Service) -> Machine:
     #     return await service.create(data)
     #
     @get(DETAIL_ROUTE)
-    async def get_machine(self, service: Service, machine_id: int) -> Machine:
+    async def get_machine(self, service: Service, machine_id: int) -> ObjectResponse[Machine]:
-        return await service.get(machine_id)
+        content = await service.get(machine_id)
+        return ObjectResponse(content=content)
+
     #
     # @put(DETAIL_ROUTE)
     # async def update_author(

app/database.py

@@ -1,6 +1,6 @@
 from typing import AsyncGenerator
 
-from litestar.contrib.sqlalchemy.plugins import SQLAlchemyAsyncConfig
+from litestar.contrib.sqlalchemy.plugins import EngineConfig, SQLAlchemyAsyncConfig
 from litestar.exceptions import ClientException
 from litestar.status_codes import HTTP_409_CONFLICT
 from sqlalchemy import URL
@@ -32,5 +32,8 @@ db_connection_string = URL.create(
     database="televend",
 )
 db_config = SQLAlchemyAsyncConfig(
-    connection_string=db_connection_string.render_as_string(hide_password=False)
+    connection_string=db_connection_string.render_as_string(hide_password=False),
+    engine_config=EngineConfig(
+        echo=True,
+    ),
 )

app/domain/machine.py

@@ -1,9 +1,14 @@
+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
@@ -11,12 +16,27 @@ from app.lib import service
 
 class Machine(BigIntBase):
     __tablename__ = "machines"
+
     caption: Mapped[str]
+    enabled: Mapped[bool]
+    alive: Mapped[bool]
+    deleted: Mapped[bool]
+    external_id: Mapped[str]
 
 
 class Repository(SQLAlchemyAsyncRepository[Machine]):
     model_type = Machine
 
+    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(Machine.alive == true())
+        return statement
+
+
 class Service(service.Service[Machine]):
     repository_type = Repository
 

app/lib/responses.py

@@ -0,0 +1,26 @@
+from dataclasses import dataclass
+from typing import Generic, TypeVar
+
+T = TypeVar("T")
+
+
+@dataclass
+class ObjectResponse(Generic[T]):
+    content: T
+
+
+@dataclass
+class ObjectListResponse(Generic[T]):
+    content: list[T]
+
+
+@dataclass
+class PaginationMeta:
+    page: int
+    page_count: int
+
+
+@dataclass
+class PaginatedObjectListResponse(Generic[T]):
+    content: list[T]
+    meta: PaginationMeta

main.py

@@ -1,56 +1,30 @@
-from uuid import UUID
+from typing import Any
 
 from litestar import Litestar, get
-from litestar.contrib.repository import FilterTypes
 from litestar.contrib.repository.exceptions import (
     RepositoryError as RepositoryException,
 )
-from litestar.contrib.repository.filters import (
-    BeforeAfter,
-    CollectionFilter,
-    LimitOffset,
-    NotInCollectionFilter,
-    NotInSearchFilter,
-    OnBeforeAfter,
-    OrderBy,
-    SearchFilter,
-)
 from litestar.contrib.sqlalchemy.plugins import SQLAlchemyPlugin
 from litestar.openapi import OpenAPIConfig
-from sqlalchemy.ext.asyncio import AsyncSession
 
 from app.controllers import create_router
 from app.database import db_config, provide_transaction
 from app.lib import exceptions
 from app.lib.service import ServiceError
 
-
+def create_app(**kwargs: Any) -> Litestar:
-@get("/")
+    return Litestar(
-async def hello_world() -> str:
+        route_handlers=[create_router()],
-    return "Hello, world!"
+        openapi_config=OpenAPIConfig(title="My API", version="1.0.0"),
+        dependencies={"session": provide_transaction},
+        plugins=[SQLAlchemyPlugin(db_config)],
+        exception_handlers={
+            RepositoryException: exceptions.repository_exception_to_http_response,  # type: ignore[dict-item]
+            ServiceError: exceptions.service_exception_to_http_response,  # type: ignore[dict-item]
+        },
+        debug=True,
+        **kwargs,
+    )
 
 
-app = Litestar(
+app = create_app()
-    route_handlers=[hello_world, create_router()],
-    openapi_config=OpenAPIConfig(title="My API", version="1.0.0"),
-    dependencies={"session": provide_transaction},
-    plugins=[SQLAlchemyPlugin(db_config)],
-    exception_handlers={
-        RepositoryException: exceptions.repository_exception_to_http_response,  # type: ignore[dict-item]
-        ServiceError: exceptions.service_exception_to_http_response,  # type: ignore[dict-item]
-    },
-    signature_namespace={
-        "AsyncSession": AsyncSession,
-        "FilterTypes": FilterTypes,
-        "BeforeAfter": BeforeAfter,
-        "CollectionFilter": CollectionFilter,
-        "LimitOffset": LimitOffset,
-        "UUID": UUID,
-        "OrderBy": OrderBy,
-        "SearchFilter": SearchFilter,
-        "OnBeforeAfter": OnBeforeAfter,
-        "NotInSearchFilter": NotInSearchFilter,
-        "NotInCollectionFilter": NotInCollectionFilter,
-    },
-    debug=True,
-)