фильтрация, пагинация, сортировка

Author: albertalexandrov

README.md

@@ -16,6 +16,7 @@
 - [Пермишены](./docs/Пермишены.md)
 - [Аутентификация](./docs/Аутентификация.md)
 - [Валидация входящих данных](./docs/Валидация%20входящих%20данных.md)
+- [Фильтрация, пагинация, сортировка.md](docs/%D0%A4%D0%B8%D0%BB%D1%8C%D1%82%D1%80%D0%B0%D1%86%D0%B8%D1%8F%2C%20%D0%BF%D0%B0%D0%B3%D0%B8%D0%BD%D0%B0%D1%86%D0%B8%D1%8F%2C%20%D1%81%D0%BE%D1%80%D1%82%D0%B8%D1%80%D0%BE%D0%B2%D0%BA%D0%B0.md)
 - [TODO](#todo)
 
 ## TODO

docs/assets/images/filtering-ordering-pagination.png

@@ -0,0 +1,138 @@
+# Фильтрация, пагинация, сортировка
+
+Предлагается объединить в одном сервисе возможности для пагинации, фильтрации, сортировки. 
+
+Это представляет собой класс, который принимает объекты запроса (Request), а также экземпляры объектов фильтрации, 
+сортировки, пагинации, которые предоставляют метод `Pagination.paginate_queryset()`, `FilterSet.filter_queryset()` и 
+`Ordering.order_queryset()` соответственно.
+
+Базовые классы сервиса и фильтрации, пагинации, сортировки:
+
+```python
+from typing import Any
+
+from fastapi import Query, Request
+from pydantic import BaseModel, Field
+
+from fastapi_django.db.repositories.queryset import QuerySet
+
+
+class Ordering(BaseModel):
+    ordering: list[str] = Field(Query(default_factory=list))
+
+    def order_queryset(self, queryset: QuerySet) -> QuerySet:
+        return queryset.order_by(*self.ordering)
+
+
+class Pagination(BaseModel):
+
+    async def paginate_queryset(self, queryset: QuerySet) -> Any:
+        raise NotImplementedError
+
+
+class LimitOffsetPagination(Pagination):
+    limit: int = Query(10, gt=0, le=100)
+    offset: int = Query(0, ge=0)
+
+    async def paginate_queryset(self, queryset: QuerySet) -> Any:
+        # пример логики пагинации
+        # пагинация знает, какие поля используются при пагинации
+        count = await queryset.count()
+        data = await queryset[self.offset:self.offset + self.limit]
+        return {"count": count, "results": data}
+
+
+class FilterSet(BaseModel):
+
+    def filter_queryset(self, queryset: QuerySet) -> QuerySet:
+        conditions = self.model_dump(exclude_unset=True, exclude_none=True)
+        return queryset.filter(**conditions)
+
+
+class ListService:
+
+    def __init__(self, request: Request | None = None, filterset=None, ordering=None, pagination=None):
+        self._request = request
+        self._filterset = filterset
+        self._ordering = ordering
+        self._pagination = pagination
+
+    async def list(self, *args, **kwargs) -> Any:
+        queryset = self.get_queryset()
+        if self._filterset is not None:
+            queryset = self._filterset.filter_queryset(queryset)
+        if self._ordering is not None:
+            queryset = self._ordering.order_queryset(queryset)
+        if self._pagination:
+            data = await self._pagination.paginate_queryset(queryset)
+        else:
+            data = await queryset
+        return data
+
+    def get_queryset(self, *args: Any, **kwargs: Any) -> QuerySet:
+        raise NotImplementedError
+
+    @classmethod
+    def init(cls, *args: Any, **kw: Any) -> Any:
+        return cls(*args, **kw)
+```
+
+Обратите внимание, что классы FilterSet, Pagination, Ordering являются наследниками pydantic.BaseModel. Это необходимо
+для валидации и отображения в свагере.
+
+Реализация:
+
+```python
+class UsersFilterSet(FilterSet):
+    # TODO: не смог сделать вложенные фильтры
+    name__ilike: str | None = Query(None)
+    role_id: int | None = Query(None)
+    role__code__in: list[str] | None = Field(Query(None, alias="role__code"))  # нужно именно прописывать Field(Query(...)), чтобы корректно отображалось в сваггере
+
+
+class UsersOrdering(Ordering):
+    ordering: list[str] = Field(Query(["id"]))  # нужно именно прописывать Field(Query(...)), чтобы корректно отображалось в сваггере
+
+
+class UsersListService(ListService):
+    # реализация сервиса, который возвращает список пользователей
+    # предоставляет возможности для фильтрации, пагинации, сортировки
+
+    def __init__(self, users: UsersRepository, **kw: Any) -> None:
+        super().__init__(**kw)
+        self._users = users
+
+    def get_queryset(self) -> QuerySet:
+        return self._users.objects.all()
+
+    @classmethod
+    def init(
+        cls,
+        request: Request,
+        users: UsersRepository = Depends(),
+        filterset: UsersFilterSet = Depends(),
+        ordering: UsersOrdering = Depends(),
+        pagination: LimitOffsetPagination = Depends(),
+    ) -> Self:
+        # метод, который использует fastapi для инициализации сервиса
+        return cls(request=request, users=users, filterset=filterset, ordering=ordering, pagination=pagination)
+```
+
+Метод UsersListService.init позволяет прописать все dependency таким образом, чтобы они корректно оторажались в сваггере.
+
+Наконец вьюха:
+
+```python
+@router.get(
+    "/filtering-ordering-paginate-users",
+    dependencies=[Depends(contextify_autocommit_session())]
+)
+async def get_users(service: UsersListService = Depends(UsersListService.init)):
+    return await service.list()
+```
+
+Сваггер:
+
+![filtering-ordering-pagination.png](assets/images/filtering-ordering-pagination.png)
+
+Пример реализации по ссылке https://github.com/albertalexandrov/fastapi-django-example/blob/main/src/web/api/crud/views.py#L60

docs/Фильтрация, пагинация, сортировка.md

@@ -2,7 +2,7 @@
 from pathlib import Path
 
 import pkg_resources
-from fastapi import APIRouter, FastAPI, Depends
+from fastapi import APIRouter, FastAPI
 from starlette.staticfiles import StaticFiles
 
 from fastapi_django.conf import settings
@@ -15,14 +15,12 @@
 
 
 def include_docs_router(app: FastAPI, router: APIRouter) -> None:
-    print("===> settings.API_DOCS_ENABLED", settings.API_DOCS_ENABLED)
     if settings.API_DOCS_ENABLED:
         app.mount(f"{settings.API_PREFIX}/static", StaticFiles(directory=APP_ROOT / "static"), name="static")
         router.include_router(docs_router)
 
 
 def setup_prometheus(app: FastAPI) -> None:
-    print("===> settings.PROMETHEUS_ENABLED", settings.PROMETHEUS_ENABLED)
     if settings.PROMETHEUS_ENABLED and "prometheus-fastapi-instrumentator" in installed_packages_list:
         from prometheus_fastapi_instrumentator import PrometheusFastApiInstrumentator
 
@@ -44,7 +42,6 @@ def include_routers(app: FastAPI) -> None:
 
 def setup_middlewares(app: FastAPI) -> None:
     for middleware in settings.MIDDLEWARES:
-        print("===> middleware", middleware)
         app.add_middleware(middleware)
     #  TODO: продумать:
     #  преднастроенные миддлварь, которые задаются в строковом формате и

fastapi_django/app.py

@@ -0,0 +1,82 @@
+from typing import Any
+
+from fastapi import Query, Request
+from pydantic import BaseModel, Field
+
+from fastapi_django.db.repositories.queryset import QuerySet
+
+
+class Ordering(BaseModel):
+    ordering: list[str] = Field(Query(default_factory=list))  # TODO: а если нужно другое именование поля?
+
+    def order_queryset(self, queryset: QuerySet) -> QuerySet:
+        return queryset.order_by(*self.ordering)
+
+
+class Pagination(BaseModel):
+
+    async def paginate_queryset(self, queryset: QuerySet) -> Any:
+        raise NotImplementedError
+
+
+class LimitOffsetPagination(Pagination):
+    limit: int = Query(10, gt=0, le=100)
+    offset: int = Query(0, ge=0)
+
+    async def paginate_queryset(self, queryset: QuerySet) -> Any:
+        # пример логики пагинации
+        # пагинация знает, какие поля используются при пагинации
+        count = await queryset.count()
+        data = await queryset[self.offset:self.offset + self.limit]
+        return {"count": count, "results": data}
+
+
+class FilterSet(BaseModel):
+
+    def filter_queryset(self, queryset: QuerySet) -> QuerySet:
+        conditions = self.model_dump(exclude_unset=True, exclude_none=True)
+        return queryset.filter(**conditions)
+
+
+class ListService:
+    # базовый класс для сервиса, возвращающего список объектов
+    # предоставляет возможности для фильтрации, пагинации, сортировки
+
+    def __init__(self, request: Request | None = None, filterset=None, ordering=None, pagination=None):
+        self._request = request
+        self._filterset = filterset
+        self._ordering = ordering
+        self._pagination = pagination
+
+    async def list(self, *args, **kwargs) -> Any:
+        queryset = self.get_queryset()
+        if self._filterset is not None:
+            queryset = self._filterset.filter_queryset(queryset)
+        if self._ordering is not None:
+            queryset = self._ordering.order_queryset(queryset)
+        if self._pagination:
+            data = await self._pagination.paginate_queryset(queryset)
+        else:
+            data = await queryset
+        return data
+
+    def get_queryset(self, *args: Any, **kwargs: Any) -> QuerySet:
+        raise NotImplementedError
+
+    @classmethod
+    def init(cls, *args: Any, **kw: Any) -> Any:
+        # метод инициалиции сервиса.  для использования в Depends
+        # предполагается, что каждый класс определяет, какие именно возможности будут использованы - фильтрация,
+        # сортировка, пагинация.  например:
+        # @classmethod
+        # def init(
+        #     cls,
+        #     request: Request,
+        #     filterset: UsersFilterSet = Depends(),
+        #     ordering: UsersOrdering = Depends(),
+        # ) -> Self:
+        #     # метод, который использует fastapi для инициализации сервиса
+        #     return cls(request=request, users=users, filterset=filterset, ordering=ordering, pagination=pagination)
+        # из примера видно, что пагинация не была определена, сделаловательно, в качестве результата можно ожидать
+        # НЕпагинированный список объекто
+        return cls(*args, **kw)

fastapi_django/db/services/__init__.py

@@ -15,3 +15,8 @@ def __init__(self, detail: str = "Forbidden", headers: dict | None = None) -> No
 class HTTP404Exception(HTTPException):
     def __init__(self, detail: str = "Not found", headers: dict | None = None) -> None:
         super().__init__(status.HTTP_404_NOT_FOUND, detail=detail, headers=headers)
+
+
+class HTTP400Exception(HTTPException):
+    def __init__(self, detail: str = "Bad request", headers: dict | None = None) -> None:
+        super().__init__(status.HTTP_400_BAD_REQUEST, detail=detail, headers=headers)

fastapi_django/db/services/list.py

@@ -7,7 +7,7 @@ check_untyped_defs = true
 
 [tool.poetry]
 name = "fastapi-django"
-version = "0.7.0"
+version = "0.7.19"
 description = ""
 authors = ["albertalexandrov <albert-ugatu@list.ru>"]
 packages = [{include = "fastapi_django"}]