feat: collection route handler generator.

src/starlite_saqlalchemy/router.py

@@ -0,0 +1,82 @@
+"""Dynamically generate routers."""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING, cast
+
+from starlite import Dependency, get
+
+from starlite_saqlalchemy.repository.filters import (
+    BeforeAfter,
+    CollectionFilter,
+    LimitOffset,
+)
+
+if TYPE_CHECKING:
+    from collections.abc import Iterable
+    from typing import Any
+
+    from pydantic import BaseModel
+    from starlite import HTTPRouteHandler
+    from typing_extensions import LiteralString
+
+    from starlite_saqlalchemy.service import Service
+
+templates = {
+    "get": "@get({params})",
+    "async_def": "async def {fn_name}({params}) -> {return_type}:",
+    "list_doc": '    """{resource} collection view."""',
+    "service_param": "service: {service_type_name}",
+    "filters_param": "filters: list[{filters_type_name}] = Dependency(skip_validation=True)",
+    "list_return": "    return [{read_dto_name}.from_orm(item) for item in await service.list(*filters)]",
+}
+
+
+def create_collection_view(
+    resource: LiteralString,
+    read_dto_type: type[BaseModel],
+    service_type: type[Service],
+    filter_types: Iterable[Any] = (BeforeAfter, CollectionFilter, LimitOffset),
+) -> HTTPRouteHandler:
+    """Create a route handler for a collection view.
+
+    Args:
+        resource: name of the domain resource, e.g., "authors"
+        read_dto_type: Pydantic model for serializing output.
+        service_type: Service object to provide the view.
+        filter_types: Collection filter types.
+
+    Returns:
+        A Starlite route handler.
+    """
+    namespace = {
+        "Dependency": Dependency,
+        "get": get,
+        read_dto_type.__name__: read_dto_type,
+        service_type.__name__: service_type,
+        **{t.__name__: t for t in filter_types},
+    }
+    params = ", ".join(
+        [
+            templates["service_param"].format(service_type_name=service_type.__name__),
+            templates["filters_param"].format(
+                filters_type_name=" | ".join(f.__name__ for f in filter_types)
+            ),
+        ]
+    )
+    fn_name = f"get_{resource}"
+    lines = [
+        templates["get"].format(params=""),
+        templates["async_def"].format(
+            fn_name=fn_name,
+            params=params,
+            return_type=f"list[{read_dto_type.__name__}]",
+        ),
+        templates["list_doc"].format(resource=resource),
+        templates["list_return"].format(read_dto_name=read_dto_type.__name__),
+    ]
+    script = "\n".join(lines)
+    eval(  # nosec B307  # noqa: SCS101  # pylint: disable=eval-used
+        compile(script, f"<generated_{resource}_{fn_name}>", "exec", dont_inherit=True),
+        namespace,
+    )
+    return cast("HTTPRouteHandler", namespace[fn_name])

tests/unit/test_router.py

@@ -0,0 +1,63 @@
+"""Tests for route generation functionality."""
+from __future__ import annotations
+
+from typing import TYPE_CHECKING
+
+import pytest
+from starlite import Provide, Starlite
+from starlite.testing import TestClient
+
+from starlite_saqlalchemy.dependencies import create_collection_dependencies
+from starlite_saqlalchemy.router import create_collection_view
+from tests.utils.domain.authors import ReadDTO, Service
+
+if TYPE_CHECKING:
+    from starlite_saqlalchemy.testing import GenericMockRepository
+
+
+@pytest.fixture(autouse=True)
+def _patch_author_service(
+    author_repository_type: GenericMockRepository,  # pylint: disable=unused-argument
+) -> None:
+    """Patch the repository for all tests."""
+
+
+@pytest.fixture(name="app")
+def fx_app(
+    author_repository_type: GenericMockRepository,  # pylint: disable=unused-argument
+) -> Starlite:
+    """Application instance with no registered handlers."""
+
+    def provide_service() -> Service:
+        """whoop."""
+        return Service(db_session=None)
+
+    dependencies = create_collection_dependencies()
+    dependencies["service"] = Provide(provide_service)
+    return Starlite(route_handlers=[], dependencies=dependencies, openapi_config=None)
+
+
+def test_create_collection_view(app: Starlite) -> None:
+    """Test collection route handler generation."""
+    handler = create_collection_view(
+        resource="authors", read_dto_type=ReadDTO, service_type=Service
+    )
+    app.register(handler)
+    with TestClient(app=app) as client:
+        resp = client.get("/")
+    assert resp.json() == [
+        {
+            "id": "97108ac1-ffcb-411d-8b1e-d9183399f63b",
+            "created": "0001-01-01T00:00:00",
+            "updated": "0001-01-01T00:00:00",
+            "name": "Agatha Christie",
+            "dob": "1890-09-15",
+        },
+        {
+            "id": "5ef29f3c-3560-4d15-ba6b-a2e5c721e4d2",
+            "created": "0001-01-01T00:00:00",
+            "updated": "0001-01-01T00:00:00",
+            "name": "Leo Tolstoy",
+            "dob": "1828-09-09",
+        },
+    ]

tests/utils/controllers.py

@@ -13,7 +13,7 @@
 DETAIL_ROUTE = "/{author_id:uuid}"
 
 
-def provides_service(db_session: AsyncSession) -> Service:
+async def provides_service(db_session: AsyncSession) -> Service:
     """Constructs repository and service objects for the request."""
     return Service(session=db_session)