mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-04-16 14:46:24 +02:00
Compare commits
3 Commits
feat/add-l
...
1a863b7032
| Author | SHA1 | Date | |
|---|---|---|---|
|
1a863b7032
|
|||
|
aca6dd298a
|
|||
|
2e7f879544
|
@@ -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`:
|
||||
|
||||
@@ -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`)"
|
||||
|
||||
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` |
|
||||
|---|---|---|
|
||||
| Total count | Yes | No |
|
||||
| Jump to arbitrary page | Yes | No |
|
||||
| Performance on deep pages | Degrades | Constant |
|
||||
| Stable under concurrent inserts | No | Yes |
|
||||
| Search compatible | Yes | Yes |
|
||||
| Use case | Admin panels, numbered pagination | Feeds, APIs, infinite scroll |
|
||||
| | `offset_paginate` | `cursor_paginate` | `paginate` |
|
||||
|---|---|---|---|
|
||||
| Return type | `OffsetPaginatedResponse` | `CursorPaginatedResponse` | either, based on `pagination_type` param |
|
||||
| Total count | Yes | No | / |
|
||||
| Jump to arbitrary page | Yes | No | / |
|
||||
| Performance on deep pages | Degrades | Constant | / |
|
||||
| Stable under concurrent inserts | No | Yes | / |
|
||||
| Use case | Admin panels, numbered pagination | Feeds, APIs, infinite scroll | single endpoint, both strategies |
|
||||
|
||||
### Offset pagination
|
||||
|
||||
```python
|
||||
@router.get(
|
||||
"",
|
||||
response_model=PaginatedResponse[User],
|
||||
)
|
||||
@router.get("")
|
||||
async def get_users(
|
||||
session: SessionDep,
|
||||
items_per_page: int = 50,
|
||||
page: int = 1,
|
||||
):
|
||||
return await crud.UserCrud.offset_paginate(
|
||||
) -> 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,
|
||||
|
||||
@@ -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)
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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,
|
||||
)
|
||||
|
||||
@@ -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",
|
||||
]
|
||||
|
||||
@@ -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,
|
||||
@@ -889,7 +896,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:
|
||||
@@ -971,7 +978,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
session, facet_fields, filters, search_joins
|
||||
)
|
||||
|
||||
return PaginatedResponse(
|
||||
return OffsetPaginatedResponse(
|
||||
data=items,
|
||||
pagination=OffsetPagination(
|
||||
total_count=total_count,
|
||||
@@ -999,7 +1006,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:
|
||||
@@ -1113,7 +1120,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
session, facet_fields, filters, search_joins
|
||||
)
|
||||
|
||||
return PaginatedResponse(
|
||||
return CursorPaginatedResponse(
|
||||
data=items,
|
||||
pagination=CursorPagination(
|
||||
next_cursor=next_cursor,
|
||||
@@ -1124,6 +1131,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],
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -2384,3 +2384,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]
|
||||
|
||||
@@ -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
|
||||
|
||||
@@ -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)."""
|
||||
|
||||
|
||||
Reference in New Issue
Block a user