Add async analog of default paginate (#1039)

Author: uriyyo

fastapi_pagination/async_paginator.py

@@ -0,0 +1,36 @@
+from typing import Any, Callable, Optional, Sequence, TypeVar
+
+__all__ = ["paginate"]
+
+from .api import apply_items_transformer, create_page
+from .bases import AbstractParams
+from .types import AdditionalData, SyncItemsTransformer
+from .utils import check_installed_extensions, verify_params
+
+T = TypeVar("T")
+
+
+# same as default paginator, but allow to use async transformer
+async def paginate(
+    sequence: Sequence[T],
+    params: Optional[AbstractParams] = None,
+    length_function: Callable[[Sequence[T]], int] = len,
+    *,
+    safe: bool = False,
+    transformer: Optional[SyncItemsTransformer] = None,
+    additional_data: Optional[AdditionalData] = None,
+) -> Any:
+    if not safe:
+        check_installed_extensions()
+
+    params, raw_params = verify_params(params, "limit-offset")
+
+    items = sequence[raw_params.as_slice()]
+    t_items = await apply_items_transformer(items, transformer, async_=True)
+
+    return create_page(
+        t_items,
+        total=length_function(sequence) if raw_params.include_total else None,
+        params=params,
+        **(additional_data or {}),
+    )

tests/test_async_paginator.py

@@ -0,0 +1,27 @@
+from fastapi import FastAPI
+from pytest import fixture
+
+from fastapi_pagination import (
+    LimitOffsetPage,
+    Page,
+    add_pagination,
+)
+from fastapi_pagination.async_paginator import paginate
+
+from .base import BasePaginationTestCase
+from .utils import OptionalLimitOffsetPage, OptionalPage
+
+
+class TestAsyncPaginationParams(BasePaginationTestCase):
+    @fixture(scope="session")
+    def app(self, model_cls, entities):
+        app = FastAPI()
+
+        @app.get("/default", response_model=Page[model_cls])
+        @app.get("/limit-offset", response_model=LimitOffsetPage[model_cls])
+        @app.get("/optional/default", response_model=OptionalPage[model_cls])
+        @app.get("/optional/limit-offset", response_model=OptionalLimitOffsetPage[model_cls])
+        async def route():
+            return await paginate(entities)
+
+        return add_pagination(app)