feat(utils): add pagination support

Author: rrbharath

README.md

@@ -1,5 +1,5 @@
 <p align="center">
-    <em>Quicker FastApi developing tools</em>
+    <em>Quicker FastAPI development tools</em>
 </p>
 <p align="center">
 <a href="https://github.com/dmontagu/fastapi-utils" target="_blank">
@@ -51,9 +51,10 @@ It also adds a variety of more basic utilities that are useful across a wide var
 * **APISettings**: A subclass of `pydantic.BaseSettings` that makes it easy to configure FastAPI through environment variables
 * **String-Valued Enums**: The `StrEnum` and `CamelStrEnum` classes make string-valued enums easier to maintain
 * **CamelCase Conversions**: Convenience functions for converting strings from `snake_case` to `camelCase` or `PascalCase` and back
+* **Pagination**: Reusable limit/offset pagination parameters and response models
 * **GUID Type**: The provided GUID type makes it easy to use UUIDs as the primary keys for your database tables
 
-See the [docs](https://fastapiutils.github.io/fastapi-utils//) for more details and examples.
+See the [docs](https://fastapiutils.github.io/fastapi-utils/) for more details and examples.
 
 ## Requirements
 

docs/index.md

@@ -51,6 +51,7 @@ It also adds a variety of more basic utilities that are useful across a wide var
 * **APISettings**: A subclass of `pydantic.BaseSettings` that makes it easy to configure FastAPI through environment variables
 * **String-Valued Enums**: The `StrEnum` and `CamelStrEnum` classes make string-valued enums easier to maintain
 * **CamelCase Conversions**: Convenience functions for converting strings from `snake_case` to `camelCase` or `PascalCase` and back
+* **Pagination**: Reusable limit/offset pagination parameters and response models
 * **GUID Type**: The provided GUID type makes it easy to use UUIDs as the primary keys for your database tables
 
 See the [docs](https://https://fastapiutils.github.io/fastapi-utils//) for more details and examples.

docs/src/pagination1.py

@@ -0,0 +1,28 @@
+from fastapi import Depends, FastAPI
+from pydantic import BaseModel
+
+from fastapi_utils.pagination import LimitOffsetParams, Page, paginate
+
+app = FastAPI()
+
+
+class UserOut(BaseModel):
+    id: int
+    name: str
+
+
+USERS = [
+    UserOut(id=1, name="Ada"),
+    UserOut(id=2, name="Grace"),
+    UserOut(id=3, name="Linus"),
+    UserOut(id=4, name="Margaret"),
+]
+
+
+@app.get("/users", response_model=Page[UserOut])
+def list_users(params: LimitOffsetParams = Depends()) -> Page[UserOut]:
+    return paginate(
+        USERS[params.offset : params.offset + params.limit],
+        total=len(USERS),
+        params=params,
+    )

docs/user-guide/basics/pagination.md

@@ -0,0 +1,33 @@
+#### Source module: [`fastapi_utils.pagination`](https://github.com/dmontagu/fastapi-utils/blob/master/fastapi_utils/pagination.py){.internal-link target=_blank}
+
+---
+
+Many APIs expose list endpoints that need the same query parameters and response shape over and over.
+The pagination utilities provide a reusable limit/offset dependency and a generic response model for those endpoints.
+
+## Limit and offset pagination
+
+Use `LimitOffsetParams` as a FastAPI dependency, then return a `Page` response with `paginate`:
+
+```python hl_lines="4 21 22"
+{!./src/pagination1.py!}
+```
+
+A request to `/users?limit=2&offset=1` returns:
+
+```JSON
+{
+  "items": [
+    {"id": 2, "name": "Grace"},
+    {"id": 3, "name": "Linus"}
+  ],
+  "total": 4,
+  "limit": 2,
+  "offset": 1,
+  "nextOffset": 3,
+  "previousOffset": 0
+}
+```
+
+The Python model fields use `snake_case`, while the generated OpenAPI schema and responses use `camelCase`,
+matching the behavior of `APIModel`.

fastapi_utils/pagination.py

@@ -0,0 +1,99 @@
+from __future__ import annotations
+
+from functools import partial
+from typing import Generic, List, Optional, Sequence, TypeVar
+
+import pydantic
+from fastapi import Query
+from pydantic import Field
+
+from .api_model import APIModel
+from .camelcase import snake2camel
+
+PYDANTIC_VERSION = pydantic.VERSION
+
+DEFAULT_LIMIT = 50
+MAX_LIMIT = 100
+
+T = TypeVar("T")
+
+
+class LimitOffsetParams:
+    """
+    A reusable FastAPI dependency for limit/offset pagination query parameters.
+    """
+
+    def __init__(
+        self,
+        limit: int = Query(DEFAULT_LIMIT, ge=1, le=MAX_LIMIT),
+        offset: int = Query(0, ge=0),
+    ):
+        if not isinstance(limit, int):
+            limit = DEFAULT_LIMIT
+        if not isinstance(offset, int):
+            offset = 0
+        self._validate(limit=limit, offset=offset)
+        self.limit = limit
+        self.offset = offset
+
+    @staticmethod
+    def _validate(*, limit: int, offset: int) -> None:
+        if limit < 1:
+            raise ValueError("limit must be greater than or equal to 1")
+        if limit > MAX_LIMIT:
+            raise ValueError(f"limit must be less than or equal to {MAX_LIMIT}")
+        if offset < 0:
+            raise ValueError("offset must be greater than or equal to 0")
+
+
+if PYDANTIC_VERSION[0] == "2":
+
+    class _PageBase(APIModel):
+        pass
+
+else:
+    from pydantic.generics import GenericModel
+
+    class _PageBase(GenericModel):
+        class Config:
+            orm_mode = True
+            allow_population_by_field_name = True
+            alias_generator = partial(snake2camel, start_lower=True)
+
+
+class Page(_PageBase, Generic[T]):
+    """
+    A generic response model for limit/offset paginated endpoints.
+    """
+
+    items: List[T]
+    total: int = Field(..., ge=0)
+    limit: int = Field(..., ge=1)
+    offset: int = Field(..., ge=0)
+    next_offset: Optional[int] = Field(None, ge=0)
+    previous_offset: Optional[int] = Field(None, ge=0)
+
+
+def paginate(items: Sequence[T], *, total: int, params: LimitOffsetParams) -> Page[T]:
+    """
+    Builds a paginated response from an already-paginated sequence and total count.
+    """
+    if total < 0:
+        raise ValueError("total must be greater than or equal to 0")
+
+    next_offset = params.offset + params.limit
+    if next_offset >= total:
+        next_offset = None
+
+    previous_offset = None
+    if params.offset > 0:
+        previous_offset = max(params.offset - params.limit, 0)
+
+    return Page(
+        items=list(items),
+        total=total,
+        limit=params.limit,
+        offset=params.offset,
+        next_offset=next_offset,
+        previous_offset=previous_offset,
+    )

mkdocs.yml

@@ -27,6 +27,7 @@ nav:
           - APISettings: "user-guide/basics/api-settings.md"
           - String-Valued Enums: "user-guide/basics/enums.md"
           - CamelCase Conversion: "user-guide/basics/camelcase.md"
+          - Pagination: "user-guide/basics/pagination.md"
           - GUID Type: "user-guide/basics/guid-type.md"
   - Get Help: "help-fastapi-utils.md"
   - Development - Contributing: "contributing.md"

tests/test_pagination.py

@@ -0,0 +1,121 @@
+from __future__ import annotations
+
+import pydantic
+import pytest
+from fastapi import Depends, FastAPI
+from starlette.testclient import TestClient
+
+from fastapi_utils.pagination import LimitOffsetParams, Page, paginate
+
+PYDANTIC_VERSION = pydantic.VERSION
+
+
+def model_dump_by_alias(model: Page[int]) -> dict[str, object]:
+    if PYDANTIC_VERSION[0] == "2":
+        return model.model_dump(by_alias=True)
+    return model.dict(by_alias=True)
+
+
+def test_paginate_first_page() -> None:
+    params = LimitOffsetParams(limit=2, offset=0)
+
+    page = paginate([1, 2], total=5, params=params)
+
+    assert page.items == [1, 2]
+    assert page.total == 5
+    assert page.limit == 2
+    assert page.offset == 0
+    assert page.next_offset == 2
+    assert page.previous_offset is None
+
+
+def test_paginate_middle_page() -> None:
+    params = LimitOffsetParams(limit=2, offset=2)
+
+    page = paginate([3, 4], total=5, params=params)
+
+    assert page.next_offset == 4
+    assert page.previous_offset == 0
+
+
+def test_paginate_last_page() -> None:
+    params = LimitOffsetParams(limit=2, offset=4)
+
+    page = paginate([5], total=5, params=params)
+
+    assert page.next_offset is None
+    assert page.previous_offset == 2
+
+
+def test_paginate_previous_offset_does_not_go_below_zero() -> None:
+    params = LimitOffsetParams(limit=50, offset=20)
+
+    page = paginate([1], total=100, params=params)
+
+    assert page.previous_offset == 0
+
+
+def test_paginate_rejects_negative_total() -> None:
+    params = LimitOffsetParams()
+
+    with pytest.raises(ValueError, match="total"):
+        paginate([], total=-1, params=params)
+
+
+def test_limit_offset_params_can_be_constructed_without_fastapi() -> None:
+    params = LimitOffsetParams()
+
+    assert params.limit == 50
+    assert params.offset == 0
+
+
+def test_limit_offset_params_validate_direct_values() -> None:
+    with pytest.raises(ValueError, match="limit"):
+        LimitOffsetParams(limit=0)
+
+    with pytest.raises(ValueError, match="limit"):
+        LimitOffsetParams(limit=101)
+
+    with pytest.raises(ValueError, match="offset"):
+        LimitOffsetParams(offset=-1)
+
+
+def test_page_uses_camel_case_aliases() -> None:
+    page = Page[int](
+        items=[1],
+        total=2,
+        limit=1,
+        offset=0,
+        next_offset=1,
+        previous_offset=None,
+    )
+
+    dumped = model_dump_by_alias(page)
+
+    assert dumped["nextOffset"] == 1
+    assert dumped["previousOffset"] is None
+
+
+def test_limit_offset_params_work_as_fastapi_dependency() -> None:
+    app = FastAPI()
+    values = [1, 2, 3, 4]
+
+    @app.get("/items", response_model=Page[int])
+    def list_items(params: LimitOffsetParams = Depends()) -> Page[int]:
+        return paginate(
+            values[params.offset : params.offset + params.limit],
+            total=len(values),
+            params=params,
+        )
+
+    response = TestClient(app).get("/items?limit=2&offset=1")
+
+    assert response.status_code == 200
+    assert response.json() == {
+        "items": [2, 3],
+        "total": 4,
+        "limit": 2,
+        "offset": 1,
+        "nextOffset": 3,
+        "previousOffset": 0,
+    }