mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-04-16 14:46:24 +02:00
Compare commits
1 Commits
1a863b7032
...
7a0b28f076
| Author | SHA1 | Date | |
|---|---|---|---|
|
7a0b28f076
|
@@ -42,17 +42,12 @@ 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:20:40"
|
```python title="routes.py:1:36"
|
||||||
--8<-- "docs_src/examples/pagination_search/routes.py:20:40"
|
--8<-- "docs_src/examples/pagination_search/routes.py:1:36"
|
||||||
```
|
```
|
||||||
|
|
||||||
**Example request**
|
**Example request**
|
||||||
@@ -66,7 +61,6 @@ 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", ... }
|
||||||
],
|
],
|
||||||
@@ -89,8 +83,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:43:63"
|
```python title="routes.py:39:59"
|
||||||
--8<-- "docs_src/examples/pagination_search/routes.py:43:63"
|
--8<-- "docs_src/examples/pagination_search/routes.py:39:59"
|
||||||
```
|
```
|
||||||
|
|
||||||
**Example request**
|
**Example request**
|
||||||
@@ -104,7 +98,6 @@ 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", ... }
|
||||||
],
|
],
|
||||||
@@ -123,47 +116,6 @@ 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`:
|
||||||
|
|||||||
@@ -145,40 +145,41 @@ 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`)"
|
||||||
|
|
||||||
Three pagination methods are available. All return a typed response whose `pagination_type` field tells clients which strategy was used.
|
Two pagination strategies are available. Both return a [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse) but differ in how they navigate through results.
|
||||||
|
|
||||||
| | `offset_paginate` | `cursor_paginate` | `paginate` |
|
| | `offset_paginate` | `cursor_paginate` |
|
||||||
|---|---|---|---|
|
|---|---|---|
|
||||||
| Return type | `OffsetPaginatedResponse` | `CursorPaginatedResponse` | either, based on `pagination_type` param |
|
| Total count | Yes | No |
|
||||||
| Total count | Yes | No | / |
|
| Jump to arbitrary page | Yes | No |
|
||||||
| Jump to arbitrary page | Yes | No | / |
|
| Performance on deep pages | Degrades | Constant |
|
||||||
| Performance on deep pages | Degrades | Constant | / |
|
| Stable under concurrent inserts | No | Yes |
|
||||||
| Stable under concurrent inserts | No | Yes | / |
|
| Search compatible | Yes | Yes |
|
||||||
| Use case | Admin panels, numbered pagination | Feeds, APIs, infinite scroll | single endpoint, both strategies |
|
| Use case | Admin panels, numbered pagination | Feeds, APIs, infinite scroll |
|
||||||
|
|
||||||
### 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 UserCrud.offset_paginate(
|
return await crud.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 an [`OffsetPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPaginatedResponse):
|
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:
|
||||||
|
|
||||||
```json
|
```json
|
||||||
{
|
{
|
||||||
"status": "SUCCESS",
|
"status": "SUCCESS",
|
||||||
"pagination_type": "offset",
|
|
||||||
"data": ["..."],
|
"data": ["..."],
|
||||||
"pagination": {
|
"pagination": {
|
||||||
"total_count": 100,
|
"total_count": 100,
|
||||||
@@ -192,26 +193,27 @@ 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 [`CursorPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPaginatedResponse):
|
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:
|
||||||
|
|
||||||
```json
|
```json
|
||||||
{
|
{
|
||||||
"status": "SUCCESS",
|
"status": "SUCCESS",
|
||||||
"pagination_type": "cursor",
|
|
||||||
"data": ["..."],
|
"data": ["..."],
|
||||||
"pagination": {
|
"pagination": {
|
||||||
"next_cursor": "eyJ2YWx1ZSI6ICIzZjQ3YWM2OS0uLi4ifQ==",
|
"next_cursor": "eyJ2YWx1ZSI6ICIzZjQ3YWM2OS0uLi4ifQ==",
|
||||||
@@ -256,41 +258,6 @@ 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).
|
||||||
@@ -333,36 +300,40 @@ 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 UserCrud.offset_paginate(
|
return await crud.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 UserCrud.cursor_paginate(
|
return await crud.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,
|
|
||||||
)
|
)
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -433,12 +404,11 @@ 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())],
|
||||||
) -> OffsetPaginatedResponse[UserRead]:
|
) -> PaginatedResponse[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,
|
|
||||||
)
|
)
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -478,8 +448,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())],
|
||||||
) -> OffsetPaginatedResponse[UserRead]:
|
) -> PaginatedResponse[UserRead]:
|
||||||
return await UserCrud.offset_paginate(session=session, order_by=order_by, schema=UserRead)
|
return await UserCrud.offset_paginate(session=session, order_by=order_by)
|
||||||
```
|
```
|
||||||
|
|
||||||
The dependency adds two query parameters to the endpoint:
|
The dependency adds two query parameters to the endpoint:
|
||||||
@@ -586,7 +556,7 @@ async def get_user(session: SessionDep, uuid: UUID) -> Response[UserRead]:
|
|||||||
)
|
)
|
||||||
|
|
||||||
@router.get("")
|
@router.get("")
|
||||||
async def list_users(session: SessionDep, page: int = 1) -> OffsetPaginatedResponse[UserRead]:
|
async def list_users(session: SessionDep, page: int = 1) -> PaginatedResponse[UserRead]:
|
||||||
return await crud.UserCrud.offset_paginate(
|
return await crud.UserCrud.offset_paginate(
|
||||||
session=session,
|
session=session,
|
||||||
page=page,
|
page=page,
|
||||||
|
|||||||
@@ -20,113 +20,50 @@ async def get_user(user: User = UserDep) -> Response[UserSchema]:
|
|||||||
return Response(data=user, message="User retrieved")
|
return Response(data=user, message="User retrieved")
|
||||||
```
|
```
|
||||||
|
|
||||||
### Paginated response models
|
### [`PaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse)
|
||||||
|
|
||||||
Three classes wrap paginated list results. Pick the one that matches your endpoint's strategy:
|
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.
|
||||||
|
|
||||||
| Class | `pagination` type | `pagination_type` field | Use when |
|
#### [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination)
|
||||||
|---|---|---|---|
|
|
||||||
| [`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 |
|
|
||||||
|
|
||||||
#### [`OffsetPaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPaginatedResponse)
|
Page-number based. Requires `total_count` so clients can compute the total number of pages.
|
||||||
|
|
||||||
!!! 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 OffsetPaginatedResponse
|
from fastapi_toolsets.schemas import PaginatedResponse, OffsetPagination
|
||||||
|
|
||||||
@router.get("/users")
|
@router.get("/users")
|
||||||
async def list_users(
|
async def list_users() -> PaginatedResponse[UserSchema]:
|
||||||
page: int = 1,
|
return PaginatedResponse(
|
||||||
items_per_page: int = 20,
|
data=users,
|
||||||
) -> OffsetPaginatedResponse[UserSchema]:
|
pagination=OffsetPagination(
|
||||||
return await UserCrud.offset_paginate(
|
total_count=100,
|
||||||
session, page=page, items_per_page=items_per_page, schema=UserSchema
|
items_per_page=10,
|
||||||
|
page=1,
|
||||||
|
has_more=True,
|
||||||
|
),
|
||||||
)
|
)
|
||||||
```
|
```
|
||||||
|
|
||||||
**Response shape:**
|
#### [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination)
|
||||||
|
|
||||||
```json
|
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.
|
||||||
{
|
|
||||||
"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 CursorPaginatedResponse
|
from fastapi_toolsets.schemas import PaginatedResponse, CursorPagination
|
||||||
|
|
||||||
@router.get("/events")
|
@router.get("/events")
|
||||||
async def list_events(
|
async def list_events() -> PaginatedResponse[EventSchema]:
|
||||||
cursor: str | None = None,
|
return PaginatedResponse(
|
||||||
items_per_page: int = 20,
|
data=events,
|
||||||
) -> CursorPaginatedResponse[EventSchema]:
|
pagination=CursorPagination(
|
||||||
return await EventCrud.cursor_paginate(
|
next_cursor="eyJpZCI6IDQyfQ==",
|
||||||
session, cursor=cursor, items_per_page=items_per_page, schema=EventSchema
|
prev_cursor=None,
|
||||||
|
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)
|
||||||
|
|||||||
@@ -14,10 +14,7 @@ from fastapi_toolsets.schemas import (
|
|||||||
ErrorResponse,
|
ErrorResponse,
|
||||||
OffsetPagination,
|
OffsetPagination,
|
||||||
CursorPagination,
|
CursorPagination,
|
||||||
PaginationType,
|
|
||||||
PaginatedResponse,
|
PaginatedResponse,
|
||||||
OffsetPaginatedResponse,
|
|
||||||
CursorPaginatedResponse,
|
|
||||||
)
|
)
|
||||||
```
|
```
|
||||||
|
|
||||||
@@ -37,10 +34,4 @@ 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
|
|
||||||
|
|||||||
@@ -72,11 +72,23 @@ async def list_articles(
|
|||||||
Depends(ArticleCrud.order_params(default_field=Article.created_at)),
|
Depends(ArticleCrud.order_params(default_field=Article.created_at)),
|
||||||
],
|
],
|
||||||
pagination_type: PaginationType = PaginationType.OFFSET,
|
pagination_type: PaginationType = PaginationType.OFFSET,
|
||||||
page: int = Query(1, ge=1),
|
page: int = Query(1, ge=1, description="Current page (offset pagination only)"),
|
||||||
cursor: str | None = None,
|
cursor: str | None = Query(
|
||||||
|
None, description="Cursor token (cursor pagination only)"
|
||||||
|
),
|
||||||
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]:
|
) -> PaginatedResponse[ArticleRead]:
|
||||||
|
"""List articles using either offset or cursor pagination.
|
||||||
|
|
||||||
|
Pass `pagination_type=offset` (default) for page-based pagination with a
|
||||||
|
total count, or `pagination_type=cursor` for efficient cursor-based
|
||||||
|
pagination suited to large datasets and infinite scroll.
|
||||||
|
|
||||||
|
- **offset**: use `page` to navigate; response includes `total_count`.
|
||||||
|
- **cursor**: use the `next_cursor` / `prev_cursor` from the previous
|
||||||
|
response as the `cursor` query parameter; no total count is returned.
|
||||||
|
"""
|
||||||
return await ArticleCrud.paginate(
|
return await ArticleCrud.paginate(
|
||||||
session,
|
session,
|
||||||
pagination_type=pagination_type,
|
pagination_type=pagination_type,
|
||||||
|
|||||||
@@ -1229,10 +1229,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
``OFFSET``, :class:`.CursorPaginatedResponse` when it is
|
``OFFSET``, :class:`.CursorPaginatedResponse` when it is
|
||||||
``CURSOR``.
|
``CURSOR``.
|
||||||
"""
|
"""
|
||||||
if items_per_page < 1:
|
if pagination_type is PaginationType.CURSOR:
|
||||||
raise ValueError(f"items_per_page must be >= 1, got {items_per_page}")
|
|
||||||
match pagination_type:
|
|
||||||
case PaginationType.CURSOR:
|
|
||||||
return await cls.cursor_paginate(
|
return await cls.cursor_paginate(
|
||||||
session,
|
session,
|
||||||
cursor=cursor,
|
cursor=cursor,
|
||||||
@@ -1248,9 +1245,6 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
filter_by=filter_by,
|
filter_by=filter_by,
|
||||||
schema=schema,
|
schema=schema,
|
||||||
)
|
)
|
||||||
case PaginationType.OFFSET:
|
|
||||||
if page < 1:
|
|
||||||
raise ValueError(f"page must be >= 1, got {page}")
|
|
||||||
return await cls.offset_paginate(
|
return await cls.offset_paginate(
|
||||||
session,
|
session,
|
||||||
filters=filters,
|
filters=filters,
|
||||||
@@ -1266,8 +1260,6 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
filter_by=filter_by,
|
filter_by=filter_by,
|
||||||
schema=schema,
|
schema=schema,
|
||||||
)
|
)
|
||||||
case _:
|
|
||||||
raise ValueError(f"Unknown pagination_type: {pagination_type!r}")
|
|
||||||
|
|
||||||
|
|
||||||
def CrudFactory(
|
def CrudFactory(
|
||||||
|
|||||||
@@ -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, PaginationType
|
from fastapi_toolsets.crud import CrudFactory
|
||||||
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,72 +2384,3 @@ 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]
|
|
||||||
|
|||||||
Reference in New Issue
Block a user