Basic users and cities api

app/controllers/__init__.py

@@ -0,0 +1,20 @@
+from litestar import Router
+
+__all__ = ["create_router"]
+
+from app.controllers.city import CityController
+from app.controllers.user import UserController
+from app.domain.city import City
+
+
+def create_router() -> Router:
+    return Router(
+        path="/v1",
+        route_handlers=[
+            CityController,
+            UserController,
+        ],
+        signature_namespace={
+            "City": City,
+        },
+    )

app/controllers/city.py

@@ -0,0 +1,63 @@
+from typing import TYPE_CHECKING, Optional
+from uuid import UUID
+
+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.city import (
+    City,
+    CityReadDTO,
+    CityWriteDTO,
+    Repository,
+    Service,
+)
+from app.lib.responses import ObjectListResponse, ObjectResponse
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+DETAIL_ROUTE = "/{city_id:uuid}"
+
+
+def provides_service(db_session: AsyncSession) -> Service:
+    """Constructs repository and service objects for the request."""
+    return Service(Repository(session=db_session))
+
+
+class CityController(Controller):
+    dto = CityWriteDTO
+    return_dto = CityReadDTO
+    path = "/cities"
+    dependencies = {
+        "service": Provide(provides_service, sync_to_thread=False),
+    }
+    tags = ["Cities"]
+
+    @get()
+    async def get_cities(
+        self, service: Service, search: Optional[str] = None
+    ) -> ObjectListResponse[City]:
+        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_city(
+        self, service: Service, city_id: UUID
+    ) -> ObjectResponse[City]:
+        content = await service.get(city_id)
+        return ObjectResponse(content=content)

app/controllers/user.py

@@ -0,0 +1,63 @@
+from typing import TYPE_CHECKING, Optional
+from uuid import UUID
+
+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.user import (
+    User,
+    UserReadDTO,
+    UserWriteDTO,
+    Repository,
+    Service,
+)
+from app.lib.responses import ObjectListResponse, ObjectResponse
+
+if TYPE_CHECKING:
+    from sqlalchemy.ext.asyncio import AsyncSession
+
+
+DETAIL_ROUTE = "/{user_id:uuid}"
+
+
+def provides_service(db_session: AsyncSession) -> Service:
+    """Constructs repository and service objects for the request."""
+    return Service(Repository(session=db_session))
+
+
+class UserController(Controller):
+    dto = UserWriteDTO
+    return_dto = UserReadDTO
+    path = "/users"
+    dependencies = {
+        "service": Provide(provides_service, sync_to_thread=False),
+    }
+    tags = ["Users"]
+
+    @get()
+    async def get_users(
+        self, service: Service, search: Optional[str] = None
+    ) -> ObjectListResponse[User]:
+        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_user(
+        self, service: Service, user_id: UUID
+    ) -> ObjectResponse[User]:
+        content = await service.get(user_id)
+        return ObjectResponse(content=content)