diff --git a/docs/examples/pagination-search.md b/docs/examples/pagination-search.md index 3b13416..5e47c5e 100644 --- a/docs/examples/pagination-search.md +++ b/docs/examples/pagination-search.md @@ -42,12 +42,17 @@ Declare `searchable_fields`, `facet_fields`, and `order_fields` once on [`CrudFa ## Routes + +```python title="routes.py:1:17" +--8<-- "docs_src/examples/pagination_search/routes.py:1:17" +``` + ### Offset pagination Best for admin panels or any UI that needs a total item count and numbered pages. -```python title="routes.py:1:36" ---8<-- "docs_src/examples/pagination_search/routes.py:1:36" +```python title="routes.py:20:40" +--8<-- "docs_src/examples/pagination_search/routes.py:20:40" ``` **Example request** @@ -61,6 +66,7 @@ GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published&or ```json { "status": "SUCCESS", + "pagination_type": "offset", "data": [ { "id": "3f47ac69-...", "title": "FastAPI tips", "status": "published", ... } ], @@ -83,8 +89,8 @@ GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published&or Best for feeds, infinite scroll, or any high-throughput API where offset performance degrades. -```python title="routes.py:39:59" ---8<-- "docs_src/examples/pagination_search/routes.py:39:59" +```python title="routes.py:43:63" +--8<-- "docs_src/examples/pagination_search/routes.py:43:63" ``` **Example request** @@ -98,6 +104,7 @@ GET /articles/cursor?items_per_page=10&status=published&order_by=created_at&orde ```json { "status": "SUCCESS", + "pagination_type": "cursor", "data": [ { "id": "3f47ac69-...", "title": "FastAPI tips", "status": "published", ... } ], @@ -116,6 +123,47 @@ GET /articles/cursor?items_per_page=10&status=published&order_by=created_at&orde Pass `next_cursor` as the `cursor` query parameter on the next request to advance to the next page. +### Unified endpoint (both strategies) + +!!! info "Added in `v2.3.0`" + +[`paginate()`](../module/crud.md#unified-paginate--both-strategies-on-one-endpoint) lets a single endpoint support both strategies via a `pagination_type` query parameter. The `pagination_type` field in the response acts as a discriminator for frontend tooling. + +```python title="routes.py:66:90" +--8<-- "docs_src/examples/pagination_search/routes.py:66:90" +``` + +**Offset request** (default) + +``` +GET /articles/?pagination_type=offset&page=1&items_per_page=10 +``` + +```json +{ + "status": "SUCCESS", + "pagination_type": "offset", + "data": ["..."], + "pagination": { "total_count": 42, "page": 1, "items_per_page": 10, "has_more": true } +} +``` + +**Cursor request** + +``` +GET /articles/?pagination_type=cursor&items_per_page=10 +GET /articles/?pagination_type=cursor&items_per_page=10&cursor=eyJ2YWx1ZSI6... +``` + +```json +{ + "status": "SUCCESS", + "pagination_type": "cursor", + "data": ["..."], + "pagination": { "next_cursor": "eyJ2YWx1ZSI6...", "prev_cursor": null, "items_per_page": 10, "has_more": true } +} +``` + ## Search behaviour Both endpoints inherit the same `searchable_fields` declared on `ArticleCrud`: diff --git a/docs/module/crud.md b/docs/module/crud.md index ea6744a..7a35105 100644 --- a/docs/module/crud.md +++ b/docs/module/crud.md @@ -145,41 +145,40 @@ user = await UserCrud.first(session=session, filters=[User.is_active == True]) !!! info "Added in `v1.1` (only offset_pagination via `paginate` if ` OffsetPaginatedResponse[UserRead]: + return await UserCrud.offset_paginate( session=session, items_per_page=items_per_page, page=page, + schema=UserRead, ) ``` -The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) method returns a [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse) whose `pagination` field is an [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination) object: +The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) method returns an [`OffsetPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPaginatedResponse): ```json { "status": "SUCCESS", + "pagination_type": "offset", "data": ["..."], "pagination": { "total_count": 100, @@ -193,27 +192,26 @@ The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.Async ### Cursor pagination ```python -@router.get( - "", - response_model=PaginatedResponse[UserRead], -) +@router.get("") async def list_users( session: SessionDep, cursor: str | None = None, items_per_page: int = 20, -): +) -> CursorPaginatedResponse[UserRead]: return await UserCrud.cursor_paginate( session=session, cursor=cursor, items_per_page=items_per_page, + schema=UserRead, ) ``` -The [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate) method returns a [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse) whose `pagination` field is a [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination) object: +The [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate) method returns a [`CursorPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPaginatedResponse): ```json { "status": "SUCCESS", + "pagination_type": "cursor", "data": ["..."], "pagination": { "next_cursor": "eyJ2YWx1ZSI6ICIzZjQ3YWM2OS0uLi4ifQ==", @@ -258,6 +256,41 @@ PostCrud = CrudFactory(model=Post, cursor_column=Post.id) PostCrud = CrudFactory(model=Post, cursor_column=Post.created_at) ``` +### Unified endpoint (both strategies) + +!!! info "Added in `v2.3.0`" + +[`paginate()`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.paginate) dispatches to `offset_paginate` or `cursor_paginate` based on a `pagination_type` query parameter, letting you expose **one endpoint** that supports both strategies. The `pagination_type` field in the response tells clients which strategy was used, enabling frontend discriminated-union typing. + +```python +from fastapi_toolsets.crud import PaginationType +from fastapi_toolsets.schemas import PaginatedResponse + +@router.get("") +async def list_users( + session: SessionDep, + pagination_type: PaginationType = PaginationType.OFFSET, + page: int = Query(1, ge=1, description="Current page (offset only)"), + cursor: str | None = Query(None, description="Cursor token (cursor only)"), + items_per_page: int = Query(20, ge=1, le=100), +) -> PaginatedResponse[UserRead]: + return await UserCrud.paginate( + session, + pagination_type=pagination_type, + page=page, + cursor=cursor, + items_per_page=items_per_page, + schema=UserRead, + ) +``` + +``` +GET /users?pagination_type=offset&page=2&items_per_page=10 +GET /users?pagination_type=cursor&cursor=eyJ2YWx1ZSI6...&items_per_page=10 +``` + +Both `page` and `cursor` are always accepted by the endpoint — unused parameters are silently ignored by `paginate()`. + ## Search Two search strategies are available, both compatible with [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) and [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate). @@ -300,40 +333,36 @@ result = await UserCrud.offset_paginate( This allows searching with both [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) and [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate): ```python -@router.get( - "", - response_model=PaginatedResponse[User], -) +@router.get("") async def get_users( session: SessionDep, items_per_page: int = 50, page: int = 1, search: str | None = None, -): - return await crud.UserCrud.offset_paginate( +) -> OffsetPaginatedResponse[UserRead]: + return await UserCrud.offset_paginate( session=session, items_per_page=items_per_page, page=page, search=search, + schema=UserRead, ) ``` ```python -@router.get( - "", - response_model=PaginatedResponse[User], -) +@router.get("") async def get_users( session: SessionDep, cursor: str | None = None, items_per_page: int = 50, search: str | None = None, -): - return await crud.UserCrud.cursor_paginate( +) -> CursorPaginatedResponse[UserRead]: + return await UserCrud.cursor_paginate( session=session, items_per_page=items_per_page, cursor=cursor, search=search, + schema=UserRead, ) ``` @@ -404,11 +433,12 @@ async def list_users( session: SessionDep, page: int = 1, filter_by: Annotated[dict[str, list[str]], Depends(UserCrud.filter_params())], -) -> PaginatedResponse[UserRead]: +) -> OffsetPaginatedResponse[UserRead]: return await UserCrud.offset_paginate( session=session, page=page, filter_by=filter_by, + schema=UserRead, ) ``` @@ -448,8 +478,8 @@ from fastapi_toolsets.crud import OrderByClause async def list_users( session: SessionDep, order_by: Annotated[OrderByClause | None, Depends(UserCrud.order_params())], -) -> PaginatedResponse[UserRead]: - return await UserCrud.offset_paginate(session=session, order_by=order_by) +) -> OffsetPaginatedResponse[UserRead]: + return await UserCrud.offset_paginate(session=session, order_by=order_by, schema=UserRead) ``` The dependency adds two query parameters to the endpoint: @@ -556,7 +586,7 @@ async def get_user(session: SessionDep, uuid: UUID) -> Response[UserRead]: ) @router.get("") -async def list_users(session: SessionDep, page: int = 1) -> PaginatedResponse[UserRead]: +async def list_users(session: SessionDep, page: int = 1) -> OffsetPaginatedResponse[UserRead]: return await crud.UserCrud.offset_paginate( session=session, page=page, diff --git a/docs/module/schemas.md b/docs/module/schemas.md index d048844..13a3fc2 100644 --- a/docs/module/schemas.md +++ b/docs/module/schemas.md @@ -20,50 +20,113 @@ async def get_user(user: User = UserDep) -> Response[UserSchema]: return Response(data=user, message="User retrieved") ``` -### [`PaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse) +### Paginated response models -Wraps a list of items with pagination metadata and optional facet values. The `pagination` field accepts either [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination) or [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination) depending on the strategy used. +Three classes wrap paginated list results. Pick the one that matches your endpoint's strategy: -#### [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination) +| Class | `pagination` type | `pagination_type` field | Use when | +|---|---|---|---| +| [`OffsetPaginatedResponse[T]`](#offsetpaginatedresponset) | `OffsetPagination` | `"offset"` (fixed) | endpoint always uses offset | +| [`CursorPaginatedResponse[T]`](#cursorpaginatedresponset) | `CursorPagination` | `"cursor"` (fixed) | endpoint always uses cursor | +| [`PaginatedResponse[T]`](#paginatedresponset) | `OffsetPagination \| CursorPagination` | — | unified endpoint supporting both strategies | -Page-number based. Requires `total_count` so clients can compute the total number of pages. +#### [`OffsetPaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPaginatedResponse) + +!!! info "Added in `v2.3.0`" + +Use as the return type when the endpoint always uses [`offset_paginate`](crud.md#offset-pagination). The `pagination` field is guaranteed to be an [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination) object; the response always includes a `pagination_type: "offset"` discriminator. ```python -from fastapi_toolsets.schemas import PaginatedResponse, OffsetPagination +from fastapi_toolsets.schemas import OffsetPaginatedResponse @router.get("/users") -async def list_users() -> PaginatedResponse[UserSchema]: - return PaginatedResponse( - data=users, - pagination=OffsetPagination( - total_count=100, - items_per_page=10, - page=1, - has_more=True, - ), +async def list_users( + page: int = 1, + items_per_page: int = 20, +) -> OffsetPaginatedResponse[UserSchema]: + return await UserCrud.offset_paginate( + session, page=page, items_per_page=items_per_page, schema=UserSchema ) ``` -#### [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination) +**Response shape:** -Cursor based. Efficient for large or frequently updated datasets where offset pagination is impractical. Provides opaque `next_cursor` / `prev_cursor` tokens; no total count is exposed. +```json +{ + "status": "SUCCESS", + "pagination_type": "offset", + "data": ["..."], + "pagination": { + "total_count": 100, + "page": 1, + "items_per_page": 20, + "has_more": true + } +} +``` + +#### [`CursorPaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPaginatedResponse) + +!!! info "Added in `v2.3.0`" + +Use as the return type when the endpoint always uses [`cursor_paginate`](crud.md#cursor-pagination). The `pagination` field is guaranteed to be a [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination) object; the response always includes a `pagination_type: "cursor"` discriminator. ```python -from fastapi_toolsets.schemas import PaginatedResponse, CursorPagination +from fastapi_toolsets.schemas import CursorPaginatedResponse @router.get("/events") -async def list_events() -> PaginatedResponse[EventSchema]: - return PaginatedResponse( - data=events, - pagination=CursorPagination( - next_cursor="eyJpZCI6IDQyfQ==", - prev_cursor=None, - items_per_page=20, - has_more=True, - ), +async def list_events( + cursor: str | None = None, + items_per_page: int = 20, +) -> CursorPaginatedResponse[EventSchema]: + return await EventCrud.cursor_paginate( + session, cursor=cursor, items_per_page=items_per_page, schema=EventSchema ) ``` +**Response shape:** + +```json +{ + "status": "SUCCESS", + "pagination_type": "cursor", + "data": ["..."], + "pagination": { + "next_cursor": "eyJpZCI6IDQyfQ==", + "prev_cursor": null, + "items_per_page": 20, + "has_more": true + } +} +``` + +#### [`PaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse) + +Base class and return type for endpoints that support **both** pagination strategies via a `pagination_type` query parameter (using [`paginate()`](crud.md#unified-paginate--both-strategies-on-one-endpoint)) + +```python +from fastapi_toolsets.crud import PaginationType +from fastapi_toolsets.schemas import PaginatedResponse + +@router.get("/users") +async def list_users( + pagination_type: PaginationType = PaginationType.OFFSET, + page: int = 1, + cursor: str | None = None, + items_per_page: int = 20, +) -> PaginatedResponse[UserSchema]: + return await UserCrud.paginate( + session, + pagination_type=pagination_type, + page=page, + cursor=cursor, + items_per_page=items_per_page, + schema=UserSchema, + ) +``` + +#### Pagination metadata models + The optional `filter_attributes` field is populated when `facet_fields` are configured on the CRUD class (see [Filter attributes](crud.md#filter-attributes-facets)). It is `None` by default and can be hidden from API responses with `response_model_exclude_none=True`. ### [`ErrorResponse`](../reference/schemas.md#fastapi_toolsets.schemas.ErrorResponse) diff --git a/docs/reference/schemas.md b/docs/reference/schemas.md index be27c87..d05b879 100644 --- a/docs/reference/schemas.md +++ b/docs/reference/schemas.md @@ -14,7 +14,10 @@ from fastapi_toolsets.schemas import ( ErrorResponse, OffsetPagination, CursorPagination, + PaginationType, PaginatedResponse, + OffsetPaginatedResponse, + CursorPaginatedResponse, ) ``` @@ -34,4 +37,10 @@ from fastapi_toolsets.schemas import ( ## ::: fastapi_toolsets.schemas.CursorPagination +## ::: fastapi_toolsets.schemas.PaginationType + ## ::: fastapi_toolsets.schemas.PaginatedResponse + +## ::: fastapi_toolsets.schemas.OffsetPaginatedResponse + +## ::: fastapi_toolsets.schemas.CursorPaginatedResponse diff --git a/docs_src/examples/pagination_search/routes.py b/docs_src/examples/pagination_search/routes.py index e88778d..bdc6857 100644 --- a/docs_src/examples/pagination_search/routes.py +++ b/docs_src/examples/pagination_search/routes.py @@ -2,8 +2,12 @@ from typing import Annotated from fastapi import APIRouter, Depends, Query -from fastapi_toolsets.crud import OrderByClause -from fastapi_toolsets.schemas import PaginatedResponse +from fastapi_toolsets.crud import OrderByClause, PaginationType +from fastapi_toolsets.schemas import ( + CursorPaginatedResponse, + OffsetPaginatedResponse, + PaginatedResponse, +) from .crud import ArticleCrud from .db import SessionDep @@ -24,7 +28,7 @@ async def list_articles_offset( page: int = Query(1, ge=1), items_per_page: int = Query(20, ge=1, le=100), search: str | None = None, -) -> PaginatedResponse[ArticleRead]: +) -> OffsetPaginatedResponse[ArticleRead]: return await ArticleCrud.offset_paginate( session=session, page=page, @@ -47,7 +51,7 @@ async def list_articles_cursor( cursor: str | None = None, items_per_page: int = Query(20, ge=1, le=100), search: str | None = None, -) -> PaginatedResponse[ArticleRead]: +) -> CursorPaginatedResponse[ArticleRead]: return await ArticleCrud.cursor_paginate( session=session, cursor=cursor, @@ -57,3 +61,30 @@ async def list_articles_cursor( order_by=order_by, schema=ArticleRead, ) + + +@router.get("/") +async def list_articles( + session: SessionDep, + filter_by: Annotated[dict[str, list[str]], Depends(ArticleCrud.filter_params())], + order_by: Annotated[ + OrderByClause | None, + Depends(ArticleCrud.order_params(default_field=Article.created_at)), + ], + pagination_type: PaginationType = PaginationType.OFFSET, + page: int = Query(1, ge=1), + cursor: str | None = None, + items_per_page: int = Query(20, ge=1, le=100), + search: str | None = None, +) -> PaginatedResponse[ArticleRead]: + return await ArticleCrud.paginate( + session, + pagination_type=pagination_type, + page=page, + cursor=cursor, + items_per_page=items_per_page, + search=search, + filter_by=filter_by or None, + order_by=order_by, + schema=ArticleRead, + ) diff --git a/src/fastapi_toolsets/crud/__init__.py b/src/fastapi_toolsets/crud/__init__.py index cb22110..bc4fb60 100644 --- a/src/fastapi_toolsets/crud/__init__.py +++ b/src/fastapi_toolsets/crud/__init__.py @@ -1,6 +1,7 @@ """Generic async CRUD operations for SQLAlchemy models.""" from ..exceptions import InvalidFacetFilterError, NoSearchableFieldsError +from ..schemas import PaginationType from ..types import ( FacetFieldType, JoinType, @@ -8,10 +9,11 @@ from ..types import ( OrderByClause, SearchFieldType, ) -from .factory import CrudFactory +from .factory import AsyncCrud, CrudFactory from .search import SearchConfig, get_searchable_fields __all__ = [ + "AsyncCrud", "CrudFactory", "FacetFieldType", "get_searchable_fields", @@ -20,6 +22,7 @@ __all__ = [ "M2MFieldType", "NoSearchableFieldsError", "OrderByClause", + "PaginationType", "SearchConfig", "SearchFieldType", ] diff --git a/src/fastapi_toolsets/crud/factory.py b/src/fastapi_toolsets/crud/factory.py index b91c88c..57c0e99 100644 --- a/src/fastapi_toolsets/crud/factory.py +++ b/src/fastapi_toolsets/crud/factory.py @@ -23,7 +23,14 @@ from sqlalchemy.sql.roles import WhereHavingRole from ..db import get_transaction from ..exceptions import InvalidOrderFieldError, NotFoundError -from ..schemas import CursorPagination, OffsetPagination, PaginatedResponse, Response +from ..schemas import ( + CursorPaginatedResponse, + CursorPagination, + OffsetPaginatedResponse, + OffsetPagination, + PaginationType, + Response, +) from ..types import ( FacetFieldType, JoinType, @@ -892,7 +899,7 @@ class AsyncCrud(Generic[ModelType]): facet_fields: Sequence[FacetFieldType] | None = None, filter_by: dict[str, Any] | BaseModel | None = None, schema: type[BaseModel], - ) -> PaginatedResponse[Any]: + ) -> OffsetPaginatedResponse[Any]: """Get paginated results using offset-based pagination. Args: @@ -974,7 +981,7 @@ class AsyncCrud(Generic[ModelType]): session, facet_fields, filters, search_joins ) - return PaginatedResponse( + return OffsetPaginatedResponse( data=items, pagination=OffsetPagination( total_count=total_count, @@ -1002,7 +1009,7 @@ class AsyncCrud(Generic[ModelType]): facet_fields: Sequence[FacetFieldType] | None = None, filter_by: dict[str, Any] | BaseModel | None = None, schema: type[BaseModel], - ) -> PaginatedResponse[Any]: + ) -> CursorPaginatedResponse[Any]: """Get paginated results using cursor-based pagination. Args: @@ -1142,7 +1149,7 @@ class AsyncCrud(Generic[ModelType]): session, facet_fields, filters, search_joins ) - return PaginatedResponse( + return CursorPaginatedResponse( data=items, pagination=CursorPagination( next_cursor=next_cursor, @@ -1153,6 +1160,144 @@ class AsyncCrud(Generic[ModelType]): filter_attributes=filter_attributes, ) + @overload + @classmethod + async def paginate( # pragma: no cover + cls: type[Self], + session: AsyncSession, + *, + pagination_type: Literal[PaginationType.OFFSET], + filters: list[Any] | None = ..., + joins: JoinType | None = ..., + outer_join: bool = ..., + load_options: Sequence[ExecutableOption] | None = ..., + order_by: OrderByClause | None = ..., + page: int = ..., + cursor: str | None = ..., + items_per_page: int = ..., + search: str | SearchConfig | None = ..., + search_fields: Sequence[SearchFieldType] | None = ..., + facet_fields: Sequence[FacetFieldType] | None = ..., + filter_by: dict[str, Any] | BaseModel | None = ..., + schema: type[BaseModel], + ) -> OffsetPaginatedResponse[Any]: ... + + @overload + @classmethod + async def paginate( # pragma: no cover + cls: type[Self], + session: AsyncSession, + *, + pagination_type: Literal[PaginationType.CURSOR], + filters: list[Any] | None = ..., + joins: JoinType | None = ..., + outer_join: bool = ..., + load_options: Sequence[ExecutableOption] | None = ..., + order_by: OrderByClause | None = ..., + page: int = ..., + cursor: str | None = ..., + items_per_page: int = ..., + search: str | SearchConfig | None = ..., + search_fields: Sequence[SearchFieldType] | None = ..., + facet_fields: Sequence[FacetFieldType] | None = ..., + filter_by: dict[str, Any] | BaseModel | None = ..., + schema: type[BaseModel], + ) -> CursorPaginatedResponse[Any]: ... + + @classmethod + async def paginate( + cls: type[Self], + session: AsyncSession, + *, + pagination_type: PaginationType = PaginationType.OFFSET, + filters: list[Any] | None = None, + joins: JoinType | None = None, + outer_join: bool = False, + load_options: Sequence[ExecutableOption] | None = None, + order_by: OrderByClause | None = None, + page: int = 1, + cursor: str | None = None, + items_per_page: int = 20, + search: str | SearchConfig | None = None, + search_fields: Sequence[SearchFieldType] | None = None, + facet_fields: Sequence[FacetFieldType] | None = None, + filter_by: dict[str, Any] | BaseModel | None = None, + schema: type[BaseModel], + ) -> OffsetPaginatedResponse[Any] | CursorPaginatedResponse[Any]: + """Get paginated results using either offset or cursor pagination. + + Args: + session: DB async session. + pagination_type: Pagination strategy. Defaults to + ``PaginationType.OFFSET``. + filters: List of SQLAlchemy filter conditions. + joins: List of ``(model, condition)`` tuples for joining related + tables. + outer_join: Use LEFT OUTER JOIN instead of INNER JOIN. + load_options: SQLAlchemy loader options. Falls back to + ``default_load_options`` when not provided. + order_by: Column or expression to order results by. + page: Page number (1-indexed). Only used when + ``pagination_type`` is ``OFFSET``. + cursor: Cursor token from a previous + :class:`.CursorPaginatedResponse`. Only used when + ``pagination_type`` is ``CURSOR``. + items_per_page: Number of items per page (default 20). + search: Search query string or :class:`.SearchConfig` object. + search_fields: Fields to search in (overrides class default). + facet_fields: Columns to compute distinct values for (overrides + class default). + filter_by: Dict of ``{column_key: value}`` to filter by declared + facet fields. Keys must match the ``column.key`` of a facet + field. Scalar → equality, list → IN clause. Raises + :exc:`.InvalidFacetFilterError` for unknown keys. + schema: Pydantic schema to serialize each item into. + + Returns: + :class:`.OffsetPaginatedResponse` when ``pagination_type`` is + ``OFFSET``, :class:`.CursorPaginatedResponse` when it is + ``CURSOR``. + """ + if items_per_page < 1: + raise ValueError(f"items_per_page must be >= 1, got {items_per_page}") + match pagination_type: + case PaginationType.CURSOR: + return await cls.cursor_paginate( + session, + cursor=cursor, + filters=filters, + joins=joins, + outer_join=outer_join, + load_options=load_options, + order_by=order_by, + items_per_page=items_per_page, + search=search, + search_fields=search_fields, + facet_fields=facet_fields, + filter_by=filter_by, + schema=schema, + ) + case PaginationType.OFFSET: + if page < 1: + raise ValueError(f"page must be >= 1, got {page}") + return await cls.offset_paginate( + session, + filters=filters, + joins=joins, + outer_join=outer_join, + load_options=load_options, + order_by=order_by, + page=page, + items_per_page=items_per_page, + search=search, + search_fields=search_fields, + facet_fields=facet_fields, + filter_by=filter_by, + schema=schema, + ) + case _: + raise ValueError(f"Unknown pagination_type: {pagination_type!r}") + def CrudFactory( model: type[ModelType], diff --git a/src/fastapi_toolsets/schemas.py b/src/fastapi_toolsets/schemas.py index 80016cd..bf0e88b 100644 --- a/src/fastapi_toolsets/schemas.py +++ b/src/fastapi_toolsets/schemas.py @@ -1,7 +1,7 @@ """Base Pydantic schemas for API responses.""" from enum import Enum -from typing import Any, ClassVar, Generic +from typing import Any, ClassVar, Generic, Literal from pydantic import BaseModel, ConfigDict @@ -10,9 +10,12 @@ from .types import DataT __all__ = [ "ApiError", "CursorPagination", + "CursorPaginatedResponse", "ErrorResponse", "OffsetPagination", + "OffsetPaginatedResponse", "PaginatedResponse", + "PaginationType", "PydanticBase", "Response", "ResponseStatus", @@ -123,9 +126,48 @@ class CursorPagination(PydanticBase): has_more: bool +class PaginationType(str, Enum): + """Pagination strategy selector for :meth:`.AsyncCrud.paginate`.""" + + OFFSET = "offset" + CURSOR = "cursor" + + class PaginatedResponse(BaseResponse, Generic[DataT]): - """Paginated API response for list endpoints.""" + """Paginated API response for list endpoints. + + Base class and return type for endpoints that support both pagination + strategies. Use :class:`OffsetPaginatedResponse` or + :class:`CursorPaginatedResponse` when the strategy is fixed; use + ``PaginatedResponse`` as the return annotation for unified endpoints that + dispatch via :meth:`~fastapi_toolsets.crud.factory.AsyncCrud.paginate`. + """ data: list[DataT] pagination: OffsetPagination | CursorPagination + pagination_type: PaginationType | None = None filter_attributes: dict[str, list[Any]] | None = None + + +class OffsetPaginatedResponse(PaginatedResponse[DataT]): + """Paginated response with typed offset-based pagination metadata. + + The ``pagination_type`` field is always ``"offset"`` and acts as a + discriminator, allowing frontend clients to narrow the union type returned + by a unified ``paginate()`` endpoint. + """ + + pagination: OffsetPagination + pagination_type: Literal[PaginationType.OFFSET] = PaginationType.OFFSET + + +class CursorPaginatedResponse(PaginatedResponse[DataT]): + """Paginated response with typed cursor-based pagination metadata. + + The ``pagination_type`` field is always ``"cursor"`` and acts as a + discriminator, allowing frontend clients to narrow the union type returned + by a unified ``paginate()`` endpoint. + """ + + pagination: CursorPagination + pagination_type: Literal[PaginationType.CURSOR] = PaginationType.CURSOR diff --git a/tests/test_crud.py b/tests/test_crud.py index 076543f..790ae72 100644 --- a/tests/test_crud.py +++ b/tests/test_crud.py @@ -6,7 +6,7 @@ import pytest from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import selectinload -from fastapi_toolsets.crud import CrudFactory +from fastapi_toolsets.crud import CrudFactory, PaginationType from fastapi_toolsets.crud.factory import AsyncCrud from fastapi_toolsets.exceptions import NotFoundError @@ -2452,3 +2452,72 @@ class TestCursorPaginateColumnTypes: page1_ids = {p.id for p in page1.data} page2_ids = {p.id for p in page2.data} assert page1_ids.isdisjoint(page2_ids) + + +class TestPaginate: + """Tests for the unified paginate() method.""" + + @pytest.mark.anyio + async def test_offset_pagination(self, db_session: AsyncSession): + """paginate() with OFFSET returns OffsetPaginatedResponse.""" + from fastapi_toolsets.schemas import OffsetPagination + + await RoleCrud.create(db_session, RoleCreate(name="admin")) + await RoleCrud.create(db_session, RoleCreate(name="user")) + + result = await RoleCrud.paginate( + db_session, + pagination_type=PaginationType.OFFSET, + schema=RoleRead, + ) + + assert isinstance(result.pagination, OffsetPagination) + assert result.pagination_type == PaginationType.OFFSET + + @pytest.mark.anyio + async def test_cursor_pagination(self, db_session: AsyncSession): + """paginate() with CURSOR returns CursorPaginatedResponse.""" + from fastapi_toolsets.schemas import CursorPagination + + await RoleCursorCrud.create(db_session, RoleCreate(name="admin")) + + result = await RoleCursorCrud.paginate( + db_session, + pagination_type=PaginationType.CURSOR, + schema=RoleRead, + ) + + assert isinstance(result.pagination, CursorPagination) + assert result.pagination_type == PaginationType.CURSOR + + @pytest.mark.anyio + async def test_invalid_items_per_page_raises(self, db_session: AsyncSession): + """paginate() raises ValueError when items_per_page < 1.""" + with pytest.raises(ValueError, match="items_per_page"): + await RoleCrud.paginate( + db_session, + pagination_type=PaginationType.OFFSET, + items_per_page=0, + schema=RoleRead, + ) + + @pytest.mark.anyio + async def test_invalid_page_raises(self, db_session: AsyncSession): + """paginate() raises ValueError when page < 1 for offset pagination.""" + with pytest.raises(ValueError, match="page"): + await RoleCrud.paginate( + db_session, + pagination_type=PaginationType.OFFSET, + page=0, + schema=RoleRead, + ) + + @pytest.mark.anyio + async def test_unknown_pagination_type_raises(self, db_session: AsyncSession): + """paginate() raises ValueError for unknown pagination_type.""" + with pytest.raises(ValueError, match="Unknown pagination_type"): + await RoleCrud.paginate( + db_session, + pagination_type="unknown", + schema=RoleRead, + ) # type: ignore[no-matching-overload] diff --git a/tests/test_example_pagination_search.py b/tests/test_example_pagination_search.py index 9ca98fe..3281cd2 100644 --- a/tests/test_example_pagination_search.py +++ b/tests/test_example_pagination_search.py @@ -393,3 +393,105 @@ class TestCursorSorting: body = resp.json() assert body["error_code"] == "SORT-422" assert body["status"] == "FAIL" + + +class TestPaginateUnified: + """Tests for the unified GET /articles/ endpoint using paginate().""" + + @pytest.mark.anyio + async def test_defaults_to_offset_pagination( + self, client: AsyncClient, ex_db_session + ): + """Without pagination_type, defaults to offset pagination.""" + await seed(ex_db_session) + + resp = await client.get("/articles/") + + assert resp.status_code == 200 + body = resp.json() + assert body["pagination_type"] == "offset" + assert "total_count" in body["pagination"] + assert body["pagination"]["total_count"] == 3 + + @pytest.mark.anyio + async def test_explicit_offset_pagination(self, client: AsyncClient, ex_db_session): + """pagination_type=offset returns OffsetPagination metadata.""" + await seed(ex_db_session) + + resp = await client.get( + "/articles/?pagination_type=offset&page=1&items_per_page=2" + ) + + assert resp.status_code == 200 + body = resp.json() + assert body["pagination_type"] == "offset" + assert body["pagination"]["total_count"] == 3 + assert body["pagination"]["page"] == 1 + assert body["pagination"]["has_more"] is True + assert len(body["data"]) == 2 + + @pytest.mark.anyio + async def test_cursor_pagination_type(self, client: AsyncClient, ex_db_session): + """pagination_type=cursor returns CursorPagination metadata.""" + await seed(ex_db_session) + + resp = await client.get("/articles/?pagination_type=cursor&items_per_page=2") + + assert resp.status_code == 200 + body = resp.json() + assert body["pagination_type"] == "cursor" + assert "next_cursor" in body["pagination"] + assert "total_count" not in body["pagination"] + assert body["pagination"]["has_more"] is True + assert len(body["data"]) == 2 + + @pytest.mark.anyio + async def test_cursor_pagination_navigate_pages( + self, client: AsyncClient, ex_db_session + ): + """Cursor from first page can be used to fetch the next page.""" + await seed(ex_db_session) + + first = await client.get("/articles/?pagination_type=cursor&items_per_page=2") + assert first.status_code == 200 + first_body = first.json() + next_cursor = first_body["pagination"]["next_cursor"] + assert next_cursor is not None + + second = await client.get( + f"/articles/?pagination_type=cursor&items_per_page=2&cursor={next_cursor}" + ) + assert second.status_code == 200 + second_body = second.json() + assert second_body["pagination_type"] == "cursor" + assert second_body["pagination"]["has_more"] is False + assert len(second_body["data"]) == 1 + + @pytest.mark.anyio + async def test_cursor_pagination_with_search( + self, client: AsyncClient, ex_db_session + ): + """paginate() with cursor type respects search parameter.""" + await seed(ex_db_session) + + resp = await client.get("/articles/?pagination_type=cursor&search=fastapi") + + assert resp.status_code == 200 + body = resp.json() + assert body["pagination_type"] == "cursor" + assert len(body["data"]) == 1 + assert body["data"][0]["title"] == "FastAPI tips" + + @pytest.mark.anyio + async def test_offset_pagination_with_filter( + self, client: AsyncClient, ex_db_session + ): + """paginate() with offset type respects filter_by parameter.""" + await seed(ex_db_session) + + resp = await client.get("/articles/?pagination_type=offset&status=published") + + assert resp.status_code == 200 + body = resp.json() + assert body["pagination_type"] == "offset" + assert body["pagination"]["total_count"] == 2 diff --git a/tests/test_schemas.py b/tests/test_schemas.py index 17694b3..a5f17a0 100644 --- a/tests/test_schemas.py +++ b/tests/test_schemas.py @@ -6,9 +6,12 @@ from pydantic import ValidationError from fastapi_toolsets.schemas import ( ApiError, CursorPagination, + CursorPaginatedResponse, ErrorResponse, OffsetPagination, + OffsetPaginatedResponse, PaginatedResponse, + PaginationType, Response, ResponseStatus, ) @@ -312,11 +315,6 @@ class TestPaginatedResponse: def test_generic_type_hint(self): """PaginatedResponse supports generic type hints.""" - - class UserOut: - id: int - name: str - pagination = OffsetPagination( total_count=1, items_per_page=10, @@ -371,6 +369,191 @@ class TestPaginatedResponse: assert isinstance(response.pagination, CursorPagination) +class TestPaginationType: + """Tests for PaginationType enum.""" + + def test_offset_value(self): + """OFFSET has string value 'offset'.""" + assert PaginationType.OFFSET == "offset" + assert PaginationType.OFFSET.value == "offset" + + def test_cursor_value(self): + """CURSOR has string value 'cursor'.""" + assert PaginationType.CURSOR == "cursor" + assert PaginationType.CURSOR.value == "cursor" + + def test_is_string_enum(self): + """PaginationType is a string enum.""" + assert isinstance(PaginationType.OFFSET, str) + assert isinstance(PaginationType.CURSOR, str) + + def test_members(self): + """PaginationType has exactly two members.""" + assert set(PaginationType) == {PaginationType.OFFSET, PaginationType.CURSOR} + + +class TestOffsetPaginatedResponse: + """Tests for OffsetPaginatedResponse schema.""" + + def test_pagination_type_is_offset(self): + """pagination_type is always PaginationType.OFFSET.""" + response = OffsetPaginatedResponse( + data=[], + pagination=OffsetPagination( + total_count=0, items_per_page=10, page=1, has_more=False + ), + ) + assert response.pagination_type is PaginationType.OFFSET + + def test_pagination_type_serializes_to_string(self): + """pagination_type serializes to 'offset' in JSON mode.""" + response = OffsetPaginatedResponse( + data=[], + pagination=OffsetPagination( + total_count=0, items_per_page=10, page=1, has_more=False + ), + ) + assert response.model_dump(mode="json")["pagination_type"] == "offset" + + def test_pagination_field_is_typed(self): + """pagination field is OffsetPagination, not the union.""" + response = OffsetPaginatedResponse( + data=[{"id": 1}], + pagination=OffsetPagination( + total_count=10, items_per_page=5, page=2, has_more=True + ), + ) + assert isinstance(response.pagination, OffsetPagination) + assert response.pagination.total_count == 10 + assert response.pagination.page == 2 + + def test_is_subclass_of_paginated_response(self): + """OffsetPaginatedResponse IS a PaginatedResponse.""" + response = OffsetPaginatedResponse( + data=[], + pagination=OffsetPagination( + total_count=0, items_per_page=10, page=1, has_more=False + ), + ) + assert isinstance(response, PaginatedResponse) + + def test_pagination_type_default_cannot_be_overridden_to_cursor(self): + """pagination_type rejects values other than OFFSET.""" + with pytest.raises(ValidationError): + OffsetPaginatedResponse( + data=[], + pagination=OffsetPagination( + total_count=0, items_per_page=10, page=1, has_more=False + ), + pagination_type=PaginationType.CURSOR, # type: ignore[arg-type] + ) + + def test_filter_attributes_defaults_to_none(self): + """filter_attributes defaults to None.""" + response = OffsetPaginatedResponse( + data=[], + pagination=OffsetPagination( + total_count=0, items_per_page=10, page=1, has_more=False + ), + ) + assert response.filter_attributes is None + + def test_full_serialization(self): + """Full JSON serialization includes all expected fields.""" + response = OffsetPaginatedResponse( + data=[{"id": 1}], + pagination=OffsetPagination( + total_count=1, items_per_page=10, page=1, has_more=False + ), + filter_attributes={"status": ["active"]}, + ) + data = response.model_dump(mode="json") + + assert data["pagination_type"] == "offset" + assert data["status"] == "SUCCESS" + assert data["data"] == [{"id": 1}] + assert data["pagination"]["total_count"] == 1 + assert data["filter_attributes"] == {"status": ["active"]} + + +class TestCursorPaginatedResponse: + """Tests for CursorPaginatedResponse schema.""" + + def test_pagination_type_is_cursor(self): + """pagination_type is always PaginationType.CURSOR.""" + response = CursorPaginatedResponse( + data=[], + pagination=CursorPagination( + next_cursor=None, items_per_page=10, has_more=False + ), + ) + assert response.pagination_type is PaginationType.CURSOR + + def test_pagination_type_serializes_to_string(self): + """pagination_type serializes to 'cursor' in JSON mode.""" + response = CursorPaginatedResponse( + data=[], + pagination=CursorPagination( + next_cursor=None, items_per_page=10, has_more=False + ), + ) + assert response.model_dump(mode="json")["pagination_type"] == "cursor" + + def test_pagination_field_is_typed(self): + """pagination field is CursorPagination, not the union.""" + response = CursorPaginatedResponse( + data=[{"id": 1}], + pagination=CursorPagination( + next_cursor="abc123", + prev_cursor=None, + items_per_page=20, + has_more=True, + ), + ) + assert isinstance(response.pagination, CursorPagination) + assert response.pagination.next_cursor == "abc123" + assert response.pagination.has_more is True + + def test_is_subclass_of_paginated_response(self): + """CursorPaginatedResponse IS a PaginatedResponse.""" + response = CursorPaginatedResponse( + data=[], + pagination=CursorPagination( + next_cursor=None, items_per_page=10, has_more=False + ), + ) + assert isinstance(response, PaginatedResponse) + + def test_pagination_type_default_cannot_be_overridden_to_offset(self): + """pagination_type rejects values other than CURSOR.""" + with pytest.raises(ValidationError): + CursorPaginatedResponse( + data=[], + pagination=CursorPagination( + next_cursor=None, items_per_page=10, has_more=False + ), + pagination_type=PaginationType.OFFSET, # type: ignore[arg-type] + ) + + def test_full_serialization(self): + """Full JSON serialization includes all expected fields.""" + response = CursorPaginatedResponse( + data=[{"id": 1}], + pagination=CursorPagination( + next_cursor="tok_next", + prev_cursor="tok_prev", + items_per_page=10, + has_more=True, + ), + ) + data = response.model_dump(mode="json") + + assert data["pagination_type"] == "cursor" + assert data["status"] == "SUCCESS" + assert data["pagination"]["next_cursor"] == "tok_next" + assert data["pagination"]["prev_cursor"] == "tok_prev" + + class TestFromAttributes: """Tests for from_attributes config (ORM mode)."""