Compare commits

...

3 Commits

Author SHA1 Message Date
1a863b7032 fix: add tests 2026-03-14 07:04:29 -04:00
aca6dd298a docs: unified paginate() endpoint 2026-03-14 06:53:28 -04:00
2e7f879544 feat: unified paginate() endpoint with typed pagination responses 2026-03-14 06:47:03 -04:00
11 changed files with 811 additions and 86 deletions

View File

@@ -42,12 +42,17 @@ Declare `searchable_fields`, `facet_fields`, and `order_fields` once on [`CrudFa
## Routes ## Routes
```python title="routes.py:1:17"
--8<-- "docs_src/examples/pagination_search/routes.py:1:17"
```
### Offset pagination ### Offset pagination
Best for admin panels or any UI that needs a total item count and numbered pages. Best for admin panels or any UI that needs a total item count and numbered pages.
```python title="routes.py:1:36" ```python title="routes.py:20:40"
--8<-- "docs_src/examples/pagination_search/routes.py:1:36" --8<-- "docs_src/examples/pagination_search/routes.py:20:40"
``` ```
**Example request** **Example request**
@@ -61,6 +66,7 @@ GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published&or
```json ```json
{ {
"status": "SUCCESS", "status": "SUCCESS",
"pagination_type": "offset",
"data": [ "data": [
{ "id": "3f47ac69-...", "title": "FastAPI tips", "status": "published", ... } { "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. Best for feeds, infinite scroll, or any high-throughput API where offset performance degrades.
```python title="routes.py:39:59" ```python title="routes.py:43:63"
--8<-- "docs_src/examples/pagination_search/routes.py:39:59" --8<-- "docs_src/examples/pagination_search/routes.py:43:63"
``` ```
**Example request** **Example request**
@@ -98,6 +104,7 @@ GET /articles/cursor?items_per_page=10&status=published&order_by=created_at&orde
```json ```json
{ {
"status": "SUCCESS", "status": "SUCCESS",
"pagination_type": "cursor",
"data": [ "data": [
{ "id": "3f47ac69-...", "title": "FastAPI tips", "status": "published", ... } { "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. 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 ## Search behaviour
Both endpoints inherit the same `searchable_fields` declared on `ArticleCrud`: Both endpoints inherit the same `searchable_fields` declared on `ArticleCrud`:

View File

@@ -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 `<v1.1`)" !!! info "Added in `v1.1` (only offset_pagination via `paginate` if `<v1.1`)"
Two pagination strategies are available. Both return a [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse) but differ in how they navigate through results. Three pagination methods are available. All return a typed response whose `pagination_type` field tells clients which strategy was used.
| | `offset_paginate` | `cursor_paginate` | | | `offset_paginate` | `cursor_paginate` | `paginate` |
|---|---|---| |---|---|---|---|
| Total count | Yes | No | | Return type | `OffsetPaginatedResponse` | `CursorPaginatedResponse` | either, based on `pagination_type` param |
| Jump to arbitrary page | Yes | No | | Total count | Yes | No | / |
| Performance on deep pages | Degrades | Constant | | Jump to arbitrary page | Yes | No | / |
| Stable under concurrent inserts | No | Yes | | Performance on deep pages | Degrades | Constant | / |
| Search compatible | Yes | Yes | | Stable under concurrent inserts | No | Yes | / |
| Use case | Admin panels, numbered pagination | Feeds, APIs, infinite scroll | | Use case | Admin panels, numbered pagination | Feeds, APIs, infinite scroll | single endpoint, both strategies |
### Offset pagination ### Offset pagination
```python ```python
@router.get( @router.get("")
"",
response_model=PaginatedResponse[User],
)
async def get_users( async def get_users(
session: SessionDep, session: SessionDep,
items_per_page: int = 50, items_per_page: int = 50,
page: int = 1, page: int = 1,
): ) -> OffsetPaginatedResponse[UserRead]:
return await crud.UserCrud.offset_paginate( return await UserCrud.offset_paginate(
session=session, session=session,
items_per_page=items_per_page, items_per_page=items_per_page,
page=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 ```json
{ {
"status": "SUCCESS", "status": "SUCCESS",
"pagination_type": "offset",
"data": ["..."], "data": ["..."],
"pagination": { "pagination": {
"total_count": 100, "total_count": 100,
@@ -193,27 +192,26 @@ The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.Async
### Cursor pagination ### Cursor pagination
```python ```python
@router.get( @router.get("")
"",
response_model=PaginatedResponse[UserRead],
)
async def list_users( async def list_users(
session: SessionDep, session: SessionDep,
cursor: str | None = None, cursor: str | None = None,
items_per_page: int = 20, items_per_page: int = 20,
): ) -> CursorPaginatedResponse[UserRead]:
return await UserCrud.cursor_paginate( return await UserCrud.cursor_paginate(
session=session, session=session,
cursor=cursor, cursor=cursor,
items_per_page=items_per_page, 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 ```json
{ {
"status": "SUCCESS", "status": "SUCCESS",
"pagination_type": "cursor",
"data": ["..."], "data": ["..."],
"pagination": { "pagination": {
"next_cursor": "eyJ2YWx1ZSI6ICIzZjQ3YWM2OS0uLi4ifQ==", "next_cursor": "eyJ2YWx1ZSI6ICIzZjQ3YWM2OS0uLi4ifQ==",
@@ -258,6 +256,41 @@ PostCrud = CrudFactory(model=Post, cursor_column=Post.id)
PostCrud = CrudFactory(model=Post, cursor_column=Post.created_at) 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 ## 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). 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): 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 ```python
@router.get( @router.get("")
"",
response_model=PaginatedResponse[User],
)
async def get_users( async def get_users(
session: SessionDep, session: SessionDep,
items_per_page: int = 50, items_per_page: int = 50,
page: int = 1, page: int = 1,
search: str | None = None, search: str | None = None,
): ) -> OffsetPaginatedResponse[UserRead]:
return await crud.UserCrud.offset_paginate( return await UserCrud.offset_paginate(
session=session, session=session,
items_per_page=items_per_page, items_per_page=items_per_page,
page=page, page=page,
search=search, search=search,
schema=UserRead,
) )
``` ```
```python ```python
@router.get( @router.get("")
"",
response_model=PaginatedResponse[User],
)
async def get_users( async def get_users(
session: SessionDep, session: SessionDep,
cursor: str | None = None, cursor: str | None = None,
items_per_page: int = 50, items_per_page: int = 50,
search: str | None = None, search: str | None = None,
): ) -> CursorPaginatedResponse[UserRead]:
return await crud.UserCrud.cursor_paginate( return await UserCrud.cursor_paginate(
session=session, session=session,
items_per_page=items_per_page, items_per_page=items_per_page,
cursor=cursor, cursor=cursor,
search=search, search=search,
schema=UserRead,
) )
``` ```
@@ -404,11 +433,12 @@ async def list_users(
session: SessionDep, session: SessionDep,
page: int = 1, page: int = 1,
filter_by: Annotated[dict[str, list[str]], Depends(UserCrud.filter_params())], filter_by: Annotated[dict[str, list[str]], Depends(UserCrud.filter_params())],
) -> PaginatedResponse[UserRead]: ) -> OffsetPaginatedResponse[UserRead]:
return await UserCrud.offset_paginate( return await UserCrud.offset_paginate(
session=session, session=session,
page=page, page=page,
filter_by=filter_by, filter_by=filter_by,
schema=UserRead,
) )
``` ```
@@ -448,8 +478,8 @@ from fastapi_toolsets.crud import OrderByClause
async def list_users( async def list_users(
session: SessionDep, session: SessionDep,
order_by: Annotated[OrderByClause | None, Depends(UserCrud.order_params())], order_by: Annotated[OrderByClause | None, Depends(UserCrud.order_params())],
) -> PaginatedResponse[UserRead]: ) -> OffsetPaginatedResponse[UserRead]:
return await UserCrud.offset_paginate(session=session, order_by=order_by) return await UserCrud.offset_paginate(session=session, order_by=order_by, schema=UserRead)
``` ```
The dependency adds two query parameters to the endpoint: 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("") @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( return await crud.UserCrud.offset_paginate(
session=session, session=session,
page=page, page=page,

View File

@@ -20,50 +20,113 @@ async def get_user(user: User = UserDep) -> Response[UserSchema]:
return Response(data=user, message="User retrieved") 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 ```python
from fastapi_toolsets.schemas import PaginatedResponse, OffsetPagination from fastapi_toolsets.schemas import OffsetPaginatedResponse
@router.get("/users") @router.get("/users")
async def list_users() -> PaginatedResponse[UserSchema]: async def list_users(
return PaginatedResponse( page: int = 1,
data=users, items_per_page: int = 20,
pagination=OffsetPagination( ) -> OffsetPaginatedResponse[UserSchema]:
total_count=100, return await UserCrud.offset_paginate(
items_per_page=10, session, page=page, items_per_page=items_per_page, schema=UserSchema
page=1,
has_more=True,
),
) )
``` ```
#### [`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 ```python
from fastapi_toolsets.schemas import PaginatedResponse, CursorPagination from fastapi_toolsets.schemas import CursorPaginatedResponse
@router.get("/events") @router.get("/events")
async def list_events() -> PaginatedResponse[EventSchema]: async def list_events(
return PaginatedResponse( cursor: str | None = None,
data=events, items_per_page: int = 20,
pagination=CursorPagination( ) -> CursorPaginatedResponse[EventSchema]:
next_cursor="eyJpZCI6IDQyfQ==", return await EventCrud.cursor_paginate(
prev_cursor=None, session, cursor=cursor, items_per_page=items_per_page, schema=EventSchema
items_per_page=20,
has_more=True,
),
) )
``` ```
**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`. 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) ### [`ErrorResponse`](../reference/schemas.md#fastapi_toolsets.schemas.ErrorResponse)

View File

@@ -14,7 +14,10 @@ from fastapi_toolsets.schemas import (
ErrorResponse, ErrorResponse,
OffsetPagination, OffsetPagination,
CursorPagination, CursorPagination,
PaginationType,
PaginatedResponse, PaginatedResponse,
OffsetPaginatedResponse,
CursorPaginatedResponse,
) )
``` ```
@@ -34,4 +37,10 @@ from fastapi_toolsets.schemas import (
## ::: fastapi_toolsets.schemas.CursorPagination ## ::: fastapi_toolsets.schemas.CursorPagination
## ::: fastapi_toolsets.schemas.PaginationType
## ::: fastapi_toolsets.schemas.PaginatedResponse ## ::: fastapi_toolsets.schemas.PaginatedResponse
## ::: fastapi_toolsets.schemas.OffsetPaginatedResponse
## ::: fastapi_toolsets.schemas.CursorPaginatedResponse

View File

@@ -2,8 +2,12 @@ from typing import Annotated
from fastapi import APIRouter, Depends, Query from fastapi import APIRouter, Depends, Query
from fastapi_toolsets.crud import OrderByClause from fastapi_toolsets.crud import OrderByClause, PaginationType
from fastapi_toolsets.schemas import PaginatedResponse from fastapi_toolsets.schemas import (
CursorPaginatedResponse,
OffsetPaginatedResponse,
PaginatedResponse,
)
from .crud import ArticleCrud from .crud import ArticleCrud
from .db import SessionDep from .db import SessionDep
@@ -24,7 +28,7 @@ async def list_articles_offset(
page: int = Query(1, ge=1), page: int = Query(1, ge=1),
items_per_page: int = Query(20, ge=1, le=100), items_per_page: int = Query(20, ge=1, le=100),
search: str | None = None, search: str | None = None,
) -> PaginatedResponse[ArticleRead]: ) -> OffsetPaginatedResponse[ArticleRead]:
return await ArticleCrud.offset_paginate( return await ArticleCrud.offset_paginate(
session=session, session=session,
page=page, page=page,
@@ -47,7 +51,7 @@ async def list_articles_cursor(
cursor: str | None = None, cursor: str | None = None,
items_per_page: int = Query(20, ge=1, le=100), items_per_page: int = Query(20, ge=1, le=100),
search: str | None = None, search: str | None = None,
) -> PaginatedResponse[ArticleRead]: ) -> CursorPaginatedResponse[ArticleRead]:
return await ArticleCrud.cursor_paginate( return await ArticleCrud.cursor_paginate(
session=session, session=session,
cursor=cursor, cursor=cursor,
@@ -57,3 +61,30 @@ async def list_articles_cursor(
order_by=order_by, order_by=order_by,
schema=ArticleRead, 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,
)

View File

@@ -1,6 +1,7 @@
"""Generic async CRUD operations for SQLAlchemy models.""" """Generic async CRUD operations for SQLAlchemy models."""
from ..exceptions import InvalidFacetFilterError, NoSearchableFieldsError from ..exceptions import InvalidFacetFilterError, NoSearchableFieldsError
from ..schemas import PaginationType
from ..types import ( from ..types import (
FacetFieldType, FacetFieldType,
JoinType, JoinType,
@@ -8,10 +9,11 @@ from ..types import (
OrderByClause, OrderByClause,
SearchFieldType, SearchFieldType,
) )
from .factory import CrudFactory from .factory import AsyncCrud, CrudFactory
from .search import SearchConfig, get_searchable_fields from .search import SearchConfig, get_searchable_fields
__all__ = [ __all__ = [
"AsyncCrud",
"CrudFactory", "CrudFactory",
"FacetFieldType", "FacetFieldType",
"get_searchable_fields", "get_searchable_fields",
@@ -20,6 +22,7 @@ __all__ = [
"M2MFieldType", "M2MFieldType",
"NoSearchableFieldsError", "NoSearchableFieldsError",
"OrderByClause", "OrderByClause",
"PaginationType",
"SearchConfig", "SearchConfig",
"SearchFieldType", "SearchFieldType",
] ]

View File

@@ -23,7 +23,14 @@ from sqlalchemy.sql.roles import WhereHavingRole
from ..db import get_transaction from ..db import get_transaction
from ..exceptions import InvalidOrderFieldError, NotFoundError from ..exceptions import InvalidOrderFieldError, NotFoundError
from ..schemas import CursorPagination, OffsetPagination, PaginatedResponse, Response from ..schemas import (
CursorPaginatedResponse,
CursorPagination,
OffsetPaginatedResponse,
OffsetPagination,
PaginationType,
Response,
)
from ..types import ( from ..types import (
FacetFieldType, FacetFieldType,
JoinType, JoinType,
@@ -889,7 +896,7 @@ class AsyncCrud(Generic[ModelType]):
facet_fields: Sequence[FacetFieldType] | None = None, facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None, filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel], schema: type[BaseModel],
) -> PaginatedResponse[Any]: ) -> OffsetPaginatedResponse[Any]:
"""Get paginated results using offset-based pagination. """Get paginated results using offset-based pagination.
Args: Args:
@@ -971,7 +978,7 @@ class AsyncCrud(Generic[ModelType]):
session, facet_fields, filters, search_joins session, facet_fields, filters, search_joins
) )
return PaginatedResponse( return OffsetPaginatedResponse(
data=items, data=items,
pagination=OffsetPagination( pagination=OffsetPagination(
total_count=total_count, total_count=total_count,
@@ -999,7 +1006,7 @@ class AsyncCrud(Generic[ModelType]):
facet_fields: Sequence[FacetFieldType] | None = None, facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None, filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel], schema: type[BaseModel],
) -> PaginatedResponse[Any]: ) -> CursorPaginatedResponse[Any]:
"""Get paginated results using cursor-based pagination. """Get paginated results using cursor-based pagination.
Args: Args:
@@ -1113,7 +1120,7 @@ class AsyncCrud(Generic[ModelType]):
session, facet_fields, filters, search_joins session, facet_fields, filters, search_joins
) )
return PaginatedResponse( return CursorPaginatedResponse(
data=items, data=items,
pagination=CursorPagination( pagination=CursorPagination(
next_cursor=next_cursor, next_cursor=next_cursor,
@@ -1124,6 +1131,144 @@ class AsyncCrud(Generic[ModelType]):
filter_attributes=filter_attributes, 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( def CrudFactory(
model: type[ModelType], model: type[ModelType],

View File

@@ -1,7 +1,7 @@
"""Base Pydantic schemas for API responses.""" """Base Pydantic schemas for API responses."""
from enum import Enum from enum import Enum
from typing import Any, ClassVar, Generic from typing import Any, ClassVar, Generic, Literal
from pydantic import BaseModel, ConfigDict from pydantic import BaseModel, ConfigDict
@@ -10,9 +10,12 @@ from .types import DataT
__all__ = [ __all__ = [
"ApiError", "ApiError",
"CursorPagination", "CursorPagination",
"CursorPaginatedResponse",
"ErrorResponse", "ErrorResponse",
"OffsetPagination", "OffsetPagination",
"OffsetPaginatedResponse",
"PaginatedResponse", "PaginatedResponse",
"PaginationType",
"PydanticBase", "PydanticBase",
"Response", "Response",
"ResponseStatus", "ResponseStatus",
@@ -123,9 +126,48 @@ class CursorPagination(PydanticBase):
has_more: bool has_more: bool
class PaginationType(str, Enum):
"""Pagination strategy selector for :meth:`.AsyncCrud.paginate`."""
OFFSET = "offset"
CURSOR = "cursor"
class PaginatedResponse(BaseResponse, Generic[DataT]): 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] data: list[DataT]
pagination: OffsetPagination | CursorPagination pagination: OffsetPagination | CursorPagination
pagination_type: PaginationType | None = None
filter_attributes: dict[str, list[Any]] | 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

View File

@@ -6,7 +6,7 @@ import pytest
from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import selectinload 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.crud.factory import AsyncCrud
from fastapi_toolsets.exceptions import NotFoundError from fastapi_toolsets.exceptions import NotFoundError
@@ -2384,3 +2384,72 @@ class TestCursorPaginateColumnTypes:
page1_ids = {p.id for p in page1.data} page1_ids = {p.id for p in page1.data}
page2_ids = {p.id for p in page2.data} page2_ids = {p.id for p in page2.data}
assert page1_ids.isdisjoint(page2_ids) 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]

View File

@@ -393,3 +393,105 @@ class TestCursorSorting:
body = resp.json() body = resp.json()
assert body["error_code"] == "SORT-422" assert body["error_code"] == "SORT-422"
assert body["status"] == "FAIL" 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

View File

@@ -6,9 +6,12 @@ from pydantic import ValidationError
from fastapi_toolsets.schemas import ( from fastapi_toolsets.schemas import (
ApiError, ApiError,
CursorPagination, CursorPagination,
CursorPaginatedResponse,
ErrorResponse, ErrorResponse,
OffsetPagination, OffsetPagination,
OffsetPaginatedResponse,
PaginatedResponse, PaginatedResponse,
PaginationType,
Response, Response,
ResponseStatus, ResponseStatus,
) )
@@ -312,11 +315,6 @@ class TestPaginatedResponse:
def test_generic_type_hint(self): def test_generic_type_hint(self):
"""PaginatedResponse supports generic type hints.""" """PaginatedResponse supports generic type hints."""
class UserOut:
id: int
name: str
pagination = OffsetPagination( pagination = OffsetPagination(
total_count=1, total_count=1,
items_per_page=10, items_per_page=10,
@@ -371,6 +369,191 @@ class TestPaginatedResponse:
assert isinstance(response.pagination, CursorPagination) 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: class TestFromAttributes:
"""Tests for from_attributes config (ORM mode).""" """Tests for from_attributes config (ORM mode)."""