Working version

app/lib/dependencies.py

@@ -0,0 +1,150 @@
+from datetime import datetime
+from uuid import UUID
+
+from litestar.contrib.repository.filters import BeforeAfter, CollectionFilter, FilterTypes, LimitOffset
+from litestar.di import Provide
+from litestar.params import Dependency, Parameter
+
+DEFAULT_PAGINATION_LIMIT = 20
+
+__all__ = [
+    "create_collection_dependencies",
+    "provide_created_filter",
+    "provide_filter_dependencies",
+    "provide_id_filter",
+    "provide_limit_offset_pagination",
+    "provide_updated_filter",
+]
+
+
+DTorNone = datetime | None
+
+CREATED_FILTER_DEPENDENCY_KEY = "created_filter"
+FILTERS_DEPENDENCY_KEY = "filters"
+ID_FILTER_DEPENDENCY_KEY = "id_filter"
+LIMIT_OFFSET_DEPENDENCY_KEY = "limit_offset"
+UPDATED_FILTER_DEPENDENCY_KEY = "updated_filter"
+
+
+def provide_id_filter(
+    ids: list[UUID] | None = Parameter(query="ids", default=None, required=False)
+) -> CollectionFilter[UUID]:
+    """Return type consumed by ``Repository.filter_in_collection()``.
+
+    Parameters
+    ----------
+    ids : list[UUID] | None
+        Parsed out of comma separated list of values in query params.
+
+    Returns:
+    -------
+    CollectionFilter[UUID]
+    """
+    return CollectionFilter(field_name="id", values=ids or [])
+
+
+def provide_created_filter(
+    before: DTorNone = Parameter(query="created-before", default=None, required=False),
+    after: DTorNone = Parameter(query="created-after", default=None, required=False),
+) -> BeforeAfter:
+    """Return type consumed by `Repository.filter_on_datetime_field()`.
+
+    Parameters
+    ----------
+    before : datetime | None
+        Filter for records updated before this date/time.
+    after : datetime | None
+        Filter for records updated after this date/time.
+    """
+    return BeforeAfter("created_at", before, after)
+
+
+def provide_updated_filter(
+    before: DTorNone = Parameter(query="updated-before", default=None, required=False),
+    after: DTorNone = Parameter(query="updated-after", default=None, required=False),
+) -> BeforeAfter:
+    """Return type consumed by `Repository.filter_on_datetime_field()`.
+    Parameters
+    ----------
+    before : datetime | None
+        Filter for records updated before this date/time.
+    after : datetime | None
+        Filter for records updated after this date/time.
+    """
+    return BeforeAfter("updated_at", before, after)
+
+
+def provide_limit_offset_pagination(
+    page: int = Parameter(ge=1, default=1, required=False),
+    page_size: int = Parameter(
+        query="page-size",
+        ge=1,
+        default=DEFAULT_PAGINATION_LIMIT,
+        required=False,
+    ),
+) -> LimitOffset:
+    """Return type consumed by `Repository.apply_limit_offset_pagination()`.
+    Parameters
+    ----------
+    page : int
+        LIMIT to apply to select.
+    page_size : int
+        OFFSET to apply to select.
+    """
+    return LimitOffset(page_size, page_size * (page - 1))
+
+
+def provide_filter_dependencies(
+    created_filter: BeforeAfter = Dependency(skip_validation=True),
+    updated_filter: BeforeAfter = Dependency(skip_validation=True),
+    id_filter: CollectionFilter = Dependency(skip_validation=True),
+    limit_offset: LimitOffset = Dependency(skip_validation=True),
+) -> list[FilterTypes]:
+    """Common collection route filtering dependencies. Add all filters to any
+    route by including this function as a dependency, e.g:
+
+        @get
+        def get_collection_handler(filters: Filters) -> ...:
+            ...
+    The dependency is provided at the application layer, so only need to inject the dependency where
+    necessary.
+
+    Parameters
+    ----------
+    id_filter : repository.CollectionFilter
+        Filter for scoping query to limited set of identities.
+    created_filter : repository.BeforeAfter
+        Filter for scoping query to instance creation date/time.
+    updated_filter : repository.BeforeAfter
+        Filter for scoping query to instance update date/time.
+    limit_offset : repository.LimitOffset
+        Filter for query pagination.
+
+    Returns:
+    -------
+    list[FilterTypes]
+        List of filters parsed from connection.
+    """
+    return [
+        created_filter,
+        id_filter,
+        limit_offset,
+        updated_filter,
+    ]
+
+
+def create_collection_dependencies() -> dict[str, Provide]:
+    """Creates a dictionary of provides for pagination endpoints.
+
+    Returns:
+    -------
+    dict[str, Provide]
+
+    """
+    return {
+        LIMIT_OFFSET_DEPENDENCY_KEY: Provide(provide_limit_offset_pagination, sync_to_thread=False),
+        UPDATED_FILTER_DEPENDENCY_KEY: Provide(provide_updated_filter, sync_to_thread=False),
+        CREATED_FILTER_DEPENDENCY_KEY: Provide(provide_created_filter, sync_to_thread=False),
+        ID_FILTER_DEPENDENCY_KEY: Provide(provide_id_filter, sync_to_thread=False),
+        FILTERS_DEPENDENCY_KEY: Provide(provide_filter_dependencies, sync_to_thread=False),
+    }