Compare commits

..

4 Commits

Author SHA1 Message Date
d3vyce
aedcbf4e04 feat: add UUIDv7Mixin (#140) 2026-03-14 20:41:46 +01:00
d3vyce
19c013bdec feat: unified paginate() endpoint with typed pagination responses (#134)
* feat: unified paginate() endpoint with typed pagination responses

* docs: unified paginate() endpoint

* fix: add tests
2026-03-14 20:23:20 +01:00
d3vyce
81407c3038 fix: use clock_timestamp() instead of now() for Mixin to ensure unique values (#138) 2026-03-14 17:30:11 +01:00
d3vyce
0fb00d44da fix: prev_cursor does not navigate backward (#136) 2026-03-14 17:18:53 +01:00
16 changed files with 1039 additions and 128 deletions

View File

@@ -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`:

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`)"
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,

View File

@@ -18,13 +18,15 @@ class Article(Base, UUIDMixin, TimestampMixin):
content: Mapped[str]
```
All timestamp columns are timezone-aware (`TIMESTAMPTZ`). All defaults are server-side, so they are also applied when inserting rows via raw SQL outside the ORM.
All timestamp columns are timezone-aware (`TIMESTAMPTZ`). All defaults are server-side (`clock_timestamp()`), so they are also applied when inserting rows via raw SQL outside the ORM.
## Mixins
### [`UUIDMixin`](../reference/models.md#fastapi_toolsets.models.UUIDMixin)
Adds a `id: UUID` primary key generated server-side by PostgreSQL using `gen_random_uuid()` (requires PostgreSQL 13+). The value is retrieved via `RETURNING` after insert, so it is available on the Python object immediately after `flush()`.
Adds a `id: UUID` primary key generated server-side by PostgreSQL using `gen_random_uuid()`. The value is retrieved via `RETURNING` after insert, so it is available on the Python object immediately after `flush()`.
!!! warning "Requires PostgreSQL 13+"
```python
from fastapi_toolsets.models import UUIDMixin
@@ -36,13 +38,37 @@ class User(Base, UUIDMixin):
# id is None before flush
user = User(username="alice")
session.add(user)
await session.flush()
print(user.id) # UUID('...')
```
### [`UUIDv7Mixin`](../reference/models.md#fastapi_toolsets.models.UUIDv7Mixin)
!!! info "Added in `v2.3`"
Adds a `id: UUID` primary key generated server-side by PostgreSQL using `uuidv7()`. It's a time-ordered UUID format that encodes a millisecond-precision timestamp in the most significant bits, making it naturally sortable and index-friendly.
!!! warning "Requires PostgreSQL 18+"
```python
from fastapi_toolsets.models import UUIDv7Mixin
class Event(Base, UUIDv7Mixin):
__tablename__ = "events"
name: Mapped[str]
# id is None before flush
event = Event(name="user.signup")
session.add(event)
await session.flush()
print(event.id) # UUID('019...')
```
### [`CreatedAtMixin`](../reference/models.md#fastapi_toolsets.models.CreatedAtMixin)
Adds a `created_at: datetime` column set to `NOW()` on insert. The column has no `onupdate` hook — it is intentionally immutable after the row is created.
Adds a `created_at: datetime` column set to `clock_timestamp()` on insert. The column has no `onupdate` hook — it is intentionally immutable after the row is created.
```python
from fastapi_toolsets.models import UUIDMixin, CreatedAtMixin
@@ -55,7 +81,7 @@ class Order(Base, UUIDMixin, CreatedAtMixin):
### [`UpdatedAtMixin`](../reference/models.md#fastapi_toolsets.models.UpdatedAtMixin)
Adds an `updated_at: datetime` column set to `NOW()` on insert and automatically updated to `NOW()` on every ORM-level update (via SQLAlchemy's `onupdate` hook).
Adds an `updated_at: datetime` column set to `clock_timestamp()` on insert and automatically updated to `clock_timestamp()` on every ORM-level update (via SQLAlchemy's `onupdate` hook).
```python
from fastapi_toolsets.models import UUIDMixin, UpdatedAtMixin

View File

@@ -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)

View File

@@ -7,6 +7,7 @@ You can import them directly from `fastapi_toolsets.models`:
```python
from fastapi_toolsets.models import (
UUIDMixin,
UUIDv7Mixin,
CreatedAtMixin,
UpdatedAtMixin,
TimestampMixin,
@@ -15,6 +16,8 @@ from fastapi_toolsets.models import (
## ::: fastapi_toolsets.models.UUIDMixin
## ::: fastapi_toolsets.models.UUIDv7Mixin
## ::: fastapi_toolsets.models.CreatedAtMixin
## ::: fastapi_toolsets.models.UpdatedAtMixin

View File

@@ -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

View File

@@ -1,9 +1,10 @@
import datetime
import uuid
from sqlalchemy import Boolean, DateTime, ForeignKey, String, Text, func
from sqlalchemy import Boolean, ForeignKey, String, Text
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from fastapi_toolsets.models import CreatedAtMixin
class Base(DeclarativeBase):
pass
@@ -18,13 +19,10 @@ class Category(Base):
articles: Mapped[list["Article"]] = relationship(back_populates="category")
class Article(Base):
class Article(Base, CreatedAtMixin):
__tablename__ = "articles"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
created_at: Mapped[datetime.datetime] = mapped_column(
DateTime(timezone=True), server_default=func.now()
)
title: Mapped[str] = mapped_column(String(256))
body: Mapped[str] = mapped_column(Text)
status: Mapped[str] = mapped_column(String(32))

View File

@@ -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,
)

View File

@@ -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",
]

View File

@@ -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,
@@ -42,14 +49,17 @@ from .search import (
)
def _encode_cursor(value: Any) -> str:
"""Encode cursor column value as an base64 string."""
return base64.b64encode(json.dumps(str(value)).encode()).decode()
def _encode_cursor(value: Any, *, direction: str = "next") -> str:
"""Encode a cursor column value and navigation direction as a base64 string."""
return base64.b64encode(
json.dumps({"val": str(value), "dir": direction}).encode()
).decode()
def _decode_cursor(cursor: str) -> str:
"""Decode cursor base64 string."""
return json.loads(base64.b64decode(cursor.encode()).decode())
def _decode_cursor(cursor: str) -> tuple[str, str]:
"""Decode a cursor base64 string into ``(raw_value, direction)``."""
payload = json.loads(base64.b64decode(cursor.encode()).decode())
return payload["val"], payload["dir"]
def _apply_joins(q: Any, joins: JoinType | None, outer_join: bool) -> Any:
@@ -889,7 +899,7 @@ class AsyncCrud(Generic[ModelType]):
facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel],
) -> PaginatedResponse[Any]:
) -> OffsetPaginatedResponse[Any]:
"""Get paginated results using offset-based pagination.
Args:
@@ -971,7 +981,7 @@ class AsyncCrud(Generic[ModelType]):
session, facet_fields, filters, search_joins
)
return PaginatedResponse(
return OffsetPaginatedResponse(
data=items,
pagination=OffsetPagination(
total_count=total_count,
@@ -999,7 +1009,7 @@ class AsyncCrud(Generic[ModelType]):
facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel],
) -> PaginatedResponse[Any]:
) -> CursorPaginatedResponse[Any]:
"""Get paginated results using cursor-based pagination.
Args:
@@ -1038,8 +1048,9 @@ class AsyncCrud(Generic[ModelType]):
cursor_column: Any = cls.cursor_column
cursor_col_name: str = cursor_column.key
direction = "next"
if cursor is not None:
raw_val = _decode_cursor(cursor)
raw_val, direction = _decode_cursor(cursor)
col_type = cursor_column.property.columns[0].type
if isinstance(col_type, Integer):
cursor_val: Any = int(raw_val)
@@ -1057,6 +1068,9 @@ class AsyncCrud(Generic[ModelType]):
"Supported types: Integer, BigInteger, SmallInteger, Uuid, "
"DateTime, Date, Float, Numeric."
)
if direction == "prev":
filters.append(cursor_column < cursor_val)
else:
filters.append(cursor_column > cursor_val)
# Build search filters
@@ -1084,12 +1098,15 @@ class AsyncCrud(Generic[ModelType]):
if resolved := cls._resolve_load_options(load_options):
q = q.options(*resolved)
# Cursor column is always the primary sort
# Cursor column is always the primary sort; reverse direction for prev traversal
if direction == "prev":
q = q.order_by(cursor_column.desc())
else:
q = q.order_by(cursor_column)
if order_by is not None:
q = q.order_by(order_by)
# Fetch one extra to detect whether a next page exists
# Fetch one extra to detect whether another page exists in this direction
q = q.limit(items_per_page + 1)
result = await session.execute(q)
raw_items = cast(list[ModelType], result.unique().scalars().all())
@@ -1097,15 +1114,34 @@ class AsyncCrud(Generic[ModelType]):
has_more = len(raw_items) > items_per_page
items_page = raw_items[:items_per_page]
# next_cursor points past the last item on this page
next_cursor: str | None = None
if has_more and items_page:
next_cursor = _encode_cursor(getattr(items_page[-1], cursor_col_name))
# Restore ascending order when traversing backward
if direction == "prev":
items_page = list(reversed(items_page))
# prev_cursor points to the first item on this page or None when on the first page
# next_cursor: points past the last item in ascending order
next_cursor: str | None = None
if direction == "next":
if has_more and items_page:
next_cursor = _encode_cursor(
getattr(items_page[-1], cursor_col_name), direction="next"
)
else:
# Going backward: always provide a next_cursor to allow returning forward
if items_page:
next_cursor = _encode_cursor(
getattr(items_page[-1], cursor_col_name), direction="next"
)
# prev_cursor: points before the first item in ascending order
prev_cursor: str | None = None
if cursor is not None and items_page:
prev_cursor = _encode_cursor(getattr(items_page[0], cursor_col_name))
if direction == "next" and cursor is not None and items_page:
prev_cursor = _encode_cursor(
getattr(items_page[0], cursor_col_name), direction="prev"
)
elif direction == "prev" and has_more and items_page:
prev_cursor = _encode_cursor(
getattr(items_page[0], cursor_col_name), direction="prev"
)
items: list[Any] = [schema.model_validate(item) for item in items_page]
@@ -1113,7 +1149,7 @@ class AsyncCrud(Generic[ModelType]):
session, facet_fields, filters, search_joins
)
return PaginatedResponse(
return CursorPaginatedResponse(
data=items,
pagination=CursorPagination(
next_cursor=next_cursor,
@@ -1124,6 +1160,144 @@ class AsyncCrud(Generic[ModelType]):
filter_attributes=filter_attributes,
)
@overload
@classmethod
async def paginate( # pragma: no cover
cls: type[Self],
session: AsyncSession,
*,
pagination_type: Literal[PaginationType.OFFSET],
filters: list[Any] | None = ...,
joins: JoinType | None = ...,
outer_join: bool = ...,
load_options: Sequence[ExecutableOption] | None = ...,
order_by: OrderByClause | None = ...,
page: int = ...,
cursor: str | None = ...,
items_per_page: int = ...,
search: str | SearchConfig | None = ...,
search_fields: Sequence[SearchFieldType] | None = ...,
facet_fields: Sequence[FacetFieldType] | None = ...,
filter_by: dict[str, Any] | BaseModel | None = ...,
schema: type[BaseModel],
) -> OffsetPaginatedResponse[Any]: ...
@overload
@classmethod
async def paginate( # pragma: no cover
cls: type[Self],
session: AsyncSession,
*,
pagination_type: Literal[PaginationType.CURSOR],
filters: list[Any] | None = ...,
joins: JoinType | None = ...,
outer_join: bool = ...,
load_options: Sequence[ExecutableOption] | None = ...,
order_by: OrderByClause | None = ...,
page: int = ...,
cursor: str | None = ...,
items_per_page: int = ...,
search: str | SearchConfig | None = ...,
search_fields: Sequence[SearchFieldType] | None = ...,
facet_fields: Sequence[FacetFieldType] | None = ...,
filter_by: dict[str, Any] | BaseModel | None = ...,
schema: type[BaseModel],
) -> CursorPaginatedResponse[Any]: ...
@classmethod
async def paginate(
cls: type[Self],
session: AsyncSession,
*,
pagination_type: PaginationType = PaginationType.OFFSET,
filters: list[Any] | None = None,
joins: JoinType | None = None,
outer_join: bool = False,
load_options: Sequence[ExecutableOption] | None = None,
order_by: OrderByClause | None = None,
page: int = 1,
cursor: str | None = None,
items_per_page: int = 20,
search: str | SearchConfig | None = None,
search_fields: Sequence[SearchFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel],
) -> OffsetPaginatedResponse[Any] | CursorPaginatedResponse[Any]:
"""Get paginated results using either offset or cursor pagination.
Args:
session: DB async session.
pagination_type: Pagination strategy. Defaults to
``PaginationType.OFFSET``.
filters: List of SQLAlchemy filter conditions.
joins: List of ``(model, condition)`` tuples for joining related
tables.
outer_join: Use LEFT OUTER JOIN instead of INNER JOIN.
load_options: SQLAlchemy loader options. Falls back to
``default_load_options`` when not provided.
order_by: Column or expression to order results by.
page: Page number (1-indexed). Only used when
``pagination_type`` is ``OFFSET``.
cursor: Cursor token from a previous
:class:`.CursorPaginatedResponse`. Only used when
``pagination_type`` is ``CURSOR``.
items_per_page: Number of items per page (default 20).
search: Search query string or :class:`.SearchConfig` object.
search_fields: Fields to search in (overrides class default).
facet_fields: Columns to compute distinct values for (overrides
class default).
filter_by: Dict of ``{column_key: value}`` to filter by declared
facet fields. Keys must match the ``column.key`` of a facet
field. Scalar → equality, list → IN clause. Raises
:exc:`.InvalidFacetFilterError` for unknown keys.
schema: Pydantic schema to serialize each item into.
Returns:
:class:`.OffsetPaginatedResponse` when ``pagination_type`` is
``OFFSET``, :class:`.CursorPaginatedResponse` when it is
``CURSOR``.
"""
if items_per_page < 1:
raise ValueError(f"items_per_page must be >= 1, got {items_per_page}")
match pagination_type:
case PaginationType.CURSOR:
return await cls.cursor_paginate(
session,
cursor=cursor,
filters=filters,
joins=joins,
outer_join=outer_join,
load_options=load_options,
order_by=order_by,
items_per_page=items_per_page,
search=search,
search_fields=search_fields,
facet_fields=facet_fields,
filter_by=filter_by,
schema=schema,
)
case PaginationType.OFFSET:
if page < 1:
raise ValueError(f"page must be >= 1, got {page}")
return await cls.offset_paginate(
session,
filters=filters,
joins=joins,
outer_join=outer_join,
load_options=load_options,
order_by=order_by,
page=page,
items_per_page=items_per_page,
search=search,
search_fields=search_fields,
facet_fields=facet_fields,
filter_by=filter_by,
schema=schema,
)
case _:
raise ValueError(f"Unknown pagination_type: {pagination_type!r}")
def CrudFactory(
model: type[ModelType],

View File

@@ -3,11 +3,12 @@
import uuid
from datetime import datetime
from sqlalchemy import DateTime, Uuid, func, text
from sqlalchemy import DateTime, Uuid, text
from sqlalchemy.orm import Mapped, mapped_column
__all__ = [
"UUIDMixin",
"UUIDv7Mixin",
"CreatedAtMixin",
"UpdatedAtMixin",
"TimestampMixin",
@@ -24,12 +25,22 @@ class UUIDMixin:
)
class UUIDv7Mixin:
"""Mixin that adds a UUIDv7 primary key auto-generated by the database."""
id: Mapped[uuid.UUID] = mapped_column(
Uuid,
primary_key=True,
server_default=text("uuidv7()"),
)
class CreatedAtMixin:
"""Mixin that adds a ``created_at`` timestamp column."""
created_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
server_default=text("clock_timestamp()"),
)
@@ -38,8 +49,8 @@ class UpdatedAtMixin:
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
server_default=text("clock_timestamp()"),
onupdate=text("clock_timestamp()"),
)

View File

@@ -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

View File

@@ -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
@@ -1969,11 +1969,8 @@ class TestCursorPaginatePrevCursor:
assert page2.pagination.prev_cursor is not None
@pytest.mark.anyio
async def test_prev_cursor_points_to_first_item(self, db_session: AsyncSession):
"""prev_cursor encodes the value of the first item on the current page."""
import base64
import json
async def test_prev_cursor_navigates_back(self, db_session: AsyncSession):
"""prev_cursor on page 2 navigates back to the same items as page 1."""
for i in range(10):
await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
@@ -1992,12 +1989,83 @@ class TestCursorPaginatePrevCursor:
assert isinstance(page2.pagination, CursorPagination)
assert page2.pagination.prev_cursor is not None
# Decode prev_cursor and compare to first item's id
decoded = json.loads(
base64.b64decode(page2.pagination.prev_cursor.encode()).decode()
# Using prev_cursor should return the same items as page 1
back_to_page1 = await RoleCursorCrud.cursor_paginate(
db_session,
cursor=page2.pagination.prev_cursor,
items_per_page=5,
schema=RoleRead,
)
first_item_id = str(page2.data[0].id)
assert decoded == first_item_id
assert isinstance(back_to_page1.pagination, CursorPagination)
assert [r.id for r in back_to_page1.data] == [r.id for r in page1.data]
assert back_to_page1.pagination.prev_cursor is None
@pytest.mark.anyio
async def test_prev_cursor_empty_result_when_no_items_before(
self, db_session: AsyncSession
):
"""Going backward past the first item returns an empty page."""
from fastapi_toolsets.crud.factory import _encode_cursor
from fastapi_toolsets.schemas import CursorPagination
await IntRoleCursorCrud.create(db_session, IntRoleCreate(name="role00"))
page1 = await IntRoleCursorCrud.cursor_paginate(
db_session, items_per_page=5, schema=IntRoleRead
)
assert isinstance(page1.pagination, CursorPagination)
# Manually craft a backward cursor before any existing id
before_all = _encode_cursor(0, direction="prev")
empty = await IntRoleCursorCrud.cursor_paginate(
db_session, cursor=before_all, items_per_page=5, schema=IntRoleRead
)
assert isinstance(empty.pagination, CursorPagination)
assert empty.data == []
assert empty.pagination.next_cursor is None
assert empty.pagination.prev_cursor is None
@pytest.mark.anyio
async def test_prev_cursor_set_when_more_pages_behind(
self, db_session: AsyncSession
):
"""Going backward on page 2 (of 3) still exposes a prev_cursor for page 1."""
for i in range(9):
await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
from fastapi_toolsets.schemas import CursorPagination
page1 = await RoleCursorCrud.cursor_paginate(
db_session, items_per_page=3, schema=RoleRead
)
assert isinstance(page1.pagination, CursorPagination)
page2 = await RoleCursorCrud.cursor_paginate(
db_session,
cursor=page1.pagination.next_cursor,
items_per_page=3,
schema=RoleRead,
)
assert isinstance(page2.pagination, CursorPagination)
page3 = await RoleCursorCrud.cursor_paginate(
db_session,
cursor=page2.pagination.next_cursor,
items_per_page=3,
schema=RoleRead,
)
assert isinstance(page3.pagination, CursorPagination)
assert page3.pagination.prev_cursor is not None
# Going back to page 2 should still have a prev_cursor pointing at page 1
back_to_page2 = await RoleCursorCrud.cursor_paginate(
db_session,
cursor=page3.pagination.prev_cursor,
items_per_page=3,
schema=RoleRead,
)
assert isinstance(back_to_page2.pagination, CursorPagination)
assert [r.id for r in back_to_page2.data] == [r.id for r in page2.data]
assert back_to_page2.pagination.prev_cursor is not None
class TestCursorPaginateWithSearch:
@@ -2384,3 +2452,72 @@ class TestCursorPaginateColumnTypes:
page1_ids = {p.id for p in page1.data}
page2_ids = {p.id for p in page2.data}
assert page1_ids.isdisjoint(page2_ids)
class TestPaginate:
"""Tests for the unified paginate() method."""
@pytest.mark.anyio
async def test_offset_pagination(self, db_session: AsyncSession):
"""paginate() with OFFSET returns OffsetPaginatedResponse."""
from fastapi_toolsets.schemas import OffsetPagination
await RoleCrud.create(db_session, RoleCreate(name="admin"))
await RoleCrud.create(db_session, RoleCreate(name="user"))
result = await RoleCrud.paginate(
db_session,
pagination_type=PaginationType.OFFSET,
schema=RoleRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination_type == PaginationType.OFFSET
@pytest.mark.anyio
async def test_cursor_pagination(self, db_session: AsyncSession):
"""paginate() with CURSOR returns CursorPaginatedResponse."""
from fastapi_toolsets.schemas import CursorPagination
await RoleCursorCrud.create(db_session, RoleCreate(name="admin"))
result = await RoleCursorCrud.paginate(
db_session,
pagination_type=PaginationType.CURSOR,
schema=RoleRead,
)
assert isinstance(result.pagination, CursorPagination)
assert result.pagination_type == PaginationType.CURSOR
@pytest.mark.anyio
async def test_invalid_items_per_page_raises(self, db_session: AsyncSession):
"""paginate() raises ValueError when items_per_page < 1."""
with pytest.raises(ValueError, match="items_per_page"):
await RoleCrud.paginate(
db_session,
pagination_type=PaginationType.OFFSET,
items_per_page=0,
schema=RoleRead,
)
@pytest.mark.anyio
async def test_invalid_page_raises(self, db_session: AsyncSession):
"""paginate() raises ValueError when page < 1 for offset pagination."""
with pytest.raises(ValueError, match="page"):
await RoleCrud.paginate(
db_session,
pagination_type=PaginationType.OFFSET,
page=0,
schema=RoleRead,
)
@pytest.mark.anyio
async def test_unknown_pagination_type_raises(self, db_session: AsyncSession):
"""paginate() raises ValueError for unknown pagination_type."""
with pytest.raises(ValueError, match="Unknown pagination_type"):
await RoleCrud.paginate(
db_session,
pagination_type="unknown",
schema=RoleRead,
) # type: ignore[no-matching-overload]

View File

@@ -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

View File

@@ -11,6 +11,7 @@ from fastapi_toolsets.models import (
CreatedAtMixin,
TimestampMixin,
UUIDMixin,
UUIDv7Mixin,
UpdatedAtMixin,
)
@@ -48,6 +49,12 @@ class TimestampModel(MixinBase, TimestampMixin):
name: Mapped[str] = mapped_column(String(50))
class UUIDv7Model(MixinBase, UUIDv7Mixin):
__tablename__ = "mixin_uuidv7_models"
name: Mapped[str] = mapped_column(String(50))
class FullMixinModel(MixinBase, UUIDMixin, UpdatedAtMixin):
__tablename__ = "mixin_full_models"
@@ -233,6 +240,50 @@ class TestTimestampMixin:
assert "updated_at" in col_names
class TestUUIDv7Mixin:
@pytest.mark.anyio
async def test_uuid7_generated_by_db(self, mixin_session):
"""UUIDv7 is generated server-side and populated after flush."""
obj = UUIDv7Model(name="test")
mixin_session.add(obj)
await mixin_session.flush()
assert obj.id is not None
assert isinstance(obj.id, uuid.UUID)
@pytest.mark.anyio
async def test_uuid7_is_primary_key(self):
"""UUIDv7Mixin adds id as primary key column."""
pk_cols = [c.name for c in UUIDv7Model.__table__.primary_key]
assert pk_cols == ["id"]
@pytest.mark.anyio
async def test_each_row_gets_unique_uuid7(self, mixin_session):
"""Each inserted row gets a distinct UUIDv7."""
a = UUIDv7Model(name="a")
b = UUIDv7Model(name="b")
mixin_session.add_all([a, b])
await mixin_session.flush()
assert a.id != b.id
@pytest.mark.anyio
async def test_uuid7_version(self, mixin_session):
"""Generated UUIDs have version 7."""
obj = UUIDv7Model(name="test")
mixin_session.add(obj)
await mixin_session.flush()
assert obj.id.version == 7
@pytest.mark.anyio
async def test_uuid7_server_default_set(self):
"""Column has uuidv7() as server default."""
col = UUIDv7Model.__table__.c["id"]
assert col.server_default is not None
assert "uuidv7" in str(col.server_default.arg)
class TestFullMixinModel:
@pytest.mark.anyio
async def test_combined_mixins_work_together(self, mixin_session):

View File

@@ -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)."""