3 Commits

Author SHA1 Message Date
44921e5966 refactor: remove deprecated parameter and function 2026-02-27 15:32:07 -05:00
117675d02f Version 1.2.1 2026-02-27 13:57:03 -05:00
d3vyce
d7ad7308c5 Add examples in documentations (#99)
* docs: fix crud

* docs: update README features

* docs: add pagination/search example

* docs: update zensical.toml

* docs: cleanup

* docs: update status to Stable + update description

* docs: add example run commands
2026-02-27 19:56:09 +01:00
24 changed files with 826 additions and 709 deletions

View File

@@ -44,7 +44,7 @@ uv add "fastapi-toolsets[all]"
### Core
- **CRUD**: Generic async CRUD operations with `CrudFactory`, built-in search with relationship traversal
- **CRUD**: Generic async CRUD operations with `CrudFactory`, built-in full-text/faceted search and Offset/Cursor pagination.
- **Database**: Session management, transaction helpers, table locking, and polling-based row change detection
- **Dependencies**: FastAPI dependency factories (`PathDependency`, `BodyDependency`) for automatic DB lookups from path or body parameters
- **Fixtures**: Fixture system with dependency management, context support, and pytest integration

View File

@@ -0,0 +1,134 @@
# Pagination & search
This example builds an articles listing endpoint that supports **offset pagination**, **cursor pagination**, **full-text search**, and **faceted filtering** — all from a single `CrudFactory` definition.
## Models
```python title="models.py"
--8<-- "docs_src/examples/pagination_search/models.py"
```
## Schemas
```python title="schemas.py"
--8<-- "docs_src/examples/pagination_search/schemas.py"
```
## Crud
Declare `facet_fields` and `searchable_fields` once on [`CrudFactory`](../reference/crud.md#fastapi_toolsets.crud.factory.CrudFactory). All endpoints built from this class share the same defaults and can override them per call.
```python title="crud.py"
--8<-- "docs_src/examples/pagination_search/crud.py"
```
## Session dependency
```python title="db.py"
--8<-- "docs_src/examples/pagination_search/db.py"
```
!!! info "Deploy a Postgres DB with docker"
```bash
docker run -d --name postgres -e POSTGRES_USER=postgres -e POSTGRES_PASSWORD=postgres -e POSTGRES_DB=postgres -p 5432:5432 postgres:18-alpine
```
## App
```python title="app.py"
--8<-- "docs_src/examples/pagination_search/app.py"
```
## Routes
### Offset pagination
Best for admin panels or any UI that needs a total item count and numbered pages.
```python title="routes.py:1:27"
--8<-- "docs_src/examples/pagination_search/routes.py:1:27"
```
**Example request**
```
GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published
```
**Example response**
```json
{
"status": "SUCCESS",
"data": [
{ "id": "3f47ac69-...", "title": "FastAPI tips", "status": "published", ... }
],
"pagination": {
"total_count": 42,
"page": 2,
"items_per_page": 10,
"has_more": true
},
"filter_attributes": {
"status": ["archived", "draft", "published"],
"name": ["backend", "frontend", "python"]
}
}
```
`filter_attributes` always reflects the values visible **after** applying the active filters. Use it to populate filter dropdowns on the client.
### Cursor pagination
Best for feeds, infinite scroll, or any high-throughput API where offset performance degrades.
```python title="routes.py:30:45"
--8<-- "docs_src/examples/pagination_search/routes.py:30:45"
```
**Example request**
```
GET /articles/cursor?items_per_page=10&status=published
```
**Example response**
```json
{
"status": "SUCCESS",
"data": [
{ "id": "3f47ac69-...", "title": "FastAPI tips", "status": "published", ... }
],
"pagination": {
"next_cursor": "eyJ2YWx1ZSI6ICIzZjQ3YWM2OS0uLi4ifQ==",
"prev_cursor": null,
"items_per_page": 10,
"has_more": true
},
"filter_attributes": {
"status": ["published"],
"name": ["backend", "python"]
}
}
```
Pass `next_cursor` as the `cursor` query parameter on the next request to advance to the next page.
## Search behaviour
Both endpoints inherit the same `searchable_fields` declared on `ArticleCrud`:
Search is **case-insensitive** and uses a `LIKE %query%` pattern. Pass a [`SearchConfig`](../reference/crud.md#fastapi_toolsets.crud.search.SearchConfig) instead of a plain string to control case sensitivity or switch to `match_mode="all"` (AND across all fields instead of OR).
```python
from fastapi_toolsets.crud import SearchConfig
# Both title AND body must contain "fastapi"
result = await ArticleCrud.offset_paginate(
session,
search=SearchConfig(query="fastapi", case_sensitive=True, match_mode="all"),
search_fields=[Article.title, Article.body],
)
```

View File

@@ -44,7 +44,7 @@ uv add "fastapi-toolsets[all]"
### Core
- **CRUD**: Generic async CRUD operations with `CrudFactory`, built-in search with relationship traversal
- **CRUD**: Generic async CRUD operations with `CrudFactory`, built-in full-text/faceted search and offset/cursor pagination.
- **Database**: Session management, transaction helpers, table locking, and polling-based row change detection
- **Dependencies**: FastAPI dependency factories (`PathDependency`, `BodyDependency`) for automatic DB lookups from path or body parameters
- **Fixtures**: Fixture system with dependency management, context support, and pytest integration

View File

@@ -95,9 +95,6 @@ The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.Async
}
```
!!! warning "Deprecated: `paginate`"
The `paginate` function is a backward-compatible alias for `offset_paginate`. This function is **deprecated** and will be removed in **v2.0**.
### Cursor pagination
```python
@@ -170,7 +167,7 @@ PostCrud = CrudFactory(model=Post, cursor_column=Post.created_at)
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).
| | Full-text search | Filter attributes |
| | Full-text search | Faceted search |
|---|---|---|
| Input | Free-text string | Exact column values |
| Relationship support | Yes | Yes |
@@ -242,7 +239,7 @@ async def get_users(
)
```
### Filter attributes
### Faceted search
!!! info "Added in `v1.2`"
@@ -384,7 +381,7 @@ await UserCrud.upsert(
)
```
## `schema` — typed response serialization
## Response serialization
!!! info "Added in `v1.1`"
@@ -417,9 +414,6 @@ async def list_users(session: SessionDep, page: int = 1) -> PaginatedResponse[Us
The schema must have `from_attributes=True` (or inherit from [`PydanticBase`](../reference/schemas.md#fastapi_toolsets.schemas.PydanticBase)) so it can be built from SQLAlchemy model instances.
!!! warning "Deprecated: `as_response`"
The `as_response=True` parameter is **deprecated** and will be removed in **v2.0**. Replace it with `schema=YourSchema`.
---
[:material-api: API Reference](../reference/crud.md)

0
docs_src/__init__.py Normal file
View File

View File

View File

@@ -0,0 +1,6 @@
from fastapi import FastAPI
from .routes import router
app = FastAPI()
app.include_router(router=router)

View File

@@ -0,0 +1,19 @@
from fastapi_toolsets.crud import CrudFactory
from .models import Article, Category
ArticleCrud = CrudFactory(
model=Article,
cursor_column=Article.created_at,
searchable_fields=[ # default fields for full-text search
Article.title,
Article.body,
(Article.category, Category.name),
],
facet_fields=[ # fields exposed as filter dropdowns
Article.status,
(Article.category, Category.name),
],
)
ArticleFilters = ArticleCrud.filter_params()

View File

@@ -0,0 +1,17 @@
from typing import Annotated
from fastapi import Depends
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from fastapi_toolsets.db import create_db_context, create_db_dependency
DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres"
engine = create_async_engine(url=DATABASE_URL, future=True)
async_session_maker = async_sessionmaker(bind=engine, expire_on_commit=False)
get_db = create_db_dependency(session_maker=async_session_maker)
get_db_context = create_db_context(session_maker=async_session_maker)
SessionDep = Annotated[AsyncSession, Depends(get_db)]

View File

@@ -0,0 +1,36 @@
import datetime
import uuid
from sqlalchemy import Boolean, DateTime, ForeignKey, String, Text, func
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
class Base(DeclarativeBase):
pass
class Category(Base):
__tablename__ = "categories"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
name: Mapped[str] = mapped_column(String(64), unique=True)
articles: Mapped[list["Article"]] = relationship(back_populates="category")
class Article(Base):
__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))
published: Mapped[bool] = mapped_column(Boolean, default=False)
category_id: Mapped[uuid.UUID | None] = mapped_column(
ForeignKey("categories.id"), nullable=True
)
category: Mapped["Category | None"] = relationship(back_populates="articles")

View File

@@ -0,0 +1,45 @@
from fastapi import APIRouter, Depends, Query
from fastapi_toolsets.schemas import PaginatedResponse
from .crud import ArticleCrud
from .db import SessionDep
from .schemas import ArticleRead
router = APIRouter(prefix="/articles")
@router.get("/offset")
async def list_articles_offset(
session: SessionDep,
page: int = Query(1, ge=1),
items_per_page: int = Query(20, ge=1, le=100),
search: str | None = None,
filter_by: dict[str, list[str]] = Depends(ArticleCrud.filter_params()),
) -> PaginatedResponse[ArticleRead]:
return await ArticleCrud.offset_paginate(
session=session,
page=page,
items_per_page=items_per_page,
search=search,
filter_by=filter_by or None,
schema=ArticleRead,
)
@router.get("/cursor")
async def list_articles_cursor(
session: SessionDep,
cursor: str | None = None,
items_per_page: int = Query(20, ge=1, le=100),
search: str | None = None,
filter_by: dict[str, list[str]] = Depends(ArticleCrud.filter_params()),
) -> PaginatedResponse[ArticleRead]:
return await ArticleCrud.cursor_paginate(
session=session,
cursor=cursor,
items_per_page=items_per_page,
search=search,
filter_by=filter_by or None,
schema=ArticleRead,
)

View File

@@ -0,0 +1,13 @@
import datetime
import uuid
from fastapi_toolsets.schemas import PydanticBase
class ArticleRead(PydanticBase):
id: uuid.UUID
created_at: datetime.datetime
title: str
status: str
published: bool
category_id: uuid.UUID | None

View File

@@ -1,7 +1,7 @@
[project]
name = "fastapi-toolsets"
version = "1.2.0"
description = "Reusable tools for FastAPI: async CRUD, fixtures, CLI, and standardized responses for SQLAlchemy + PostgreSQL"
version = "1.2.1"
description = "Production-ready utilities for FastAPI applications"
readme = "README.md"
license = "MIT"
license-files = ["LICENSE"]
@@ -11,7 +11,7 @@ authors = [
]
keywords = ["fastapi", "sqlalchemy", "postgresql"]
classifiers = [
"Development Status :: 4 - Beta",
"Development Status :: 5 - Production/Stable",
"Framework :: AsyncIO",
"Framework :: FastAPI",
"Framework :: Pydantic",

View File

@@ -21,4 +21,4 @@ Example usage:
return Response(data={"user": user.username}, message="Success")
"""
__version__ = "1.2.0"
__version__ = "1.2.1"

View File

@@ -6,7 +6,6 @@ import base64
import inspect
import json
import uuid as uuid_module
import warnings
from collections.abc import Awaitable, Callable, Mapping, Sequence
from datetime import date, datetime
from decimal import Decimal
@@ -184,10 +183,8 @@ class AsyncCrud(Generic[ModelType]):
obj: BaseModel,
*,
schema: type[SchemaType],
as_response: bool = ...,
) -> Response[SchemaType]: ...
# Backward-compatible - will be removed in v2.0
@overload
@classmethod
async def create( # pragma: no cover
@@ -195,18 +192,6 @@ class AsyncCrud(Generic[ModelType]):
session: AsyncSession,
obj: BaseModel,
*,
as_response: Literal[True],
schema: None = ...,
) -> Response[ModelType]: ...
@overload
@classmethod
async def create( # pragma: no cover
cls: type[Self],
session: AsyncSession,
obj: BaseModel,
*,
as_response: Literal[False] = ...,
schema: None = ...,
) -> ModelType: ...
@@ -216,29 +201,19 @@ class AsyncCrud(Generic[ModelType]):
session: AsyncSession,
obj: BaseModel,
*,
as_response: bool = False,
schema: type[BaseModel] | None = None,
) -> ModelType | Response[ModelType] | Response[Any]:
) -> ModelType | Response[Any]:
"""Create a new record in the database.
Args:
session: DB async session
obj: Pydantic model with data to create
as_response: Deprecated. Use ``schema`` instead. Will be removed in v2.0.
schema: Pydantic schema to serialize the result into. When provided,
the result is automatically wrapped in a ``Response[schema]``.
Returns:
Created model instance, or ``Response[schema]`` when ``schema`` is given,
or ``Response[ModelType]`` when ``as_response=True`` (deprecated).
Created model instance, or ``Response[schema]`` when ``schema`` is given.
"""
if as_response and schema is None:
warnings.warn(
"as_response is deprecated and will be removed in v2.0. "
"Use schema=YourSchema instead.",
DeprecationWarning,
stacklevel=2,
)
async with get_transaction(session):
m2m_exclude = cls._m2m_schema_fields()
data = (
@@ -254,7 +229,7 @@ class AsyncCrud(Generic[ModelType]):
session.add(db_model)
await session.refresh(db_model)
result = cast(ModelType, db_model)
if as_response or schema:
if schema:
data_out = schema.model_validate(result) if schema else result
return Response(data=data_out)
return result
@@ -271,10 +246,8 @@ class AsyncCrud(Generic[ModelType]):
with_for_update: bool = False,
load_options: list[ExecutableOption] | None = None,
schema: type[SchemaType],
as_response: bool = ...,
) -> Response[SchemaType]: ...
# Backward-compatible - will be removed in v2.0
@overload
@classmethod
async def get( # pragma: no cover
@@ -286,22 +259,6 @@ class AsyncCrud(Generic[ModelType]):
outer_join: bool = False,
with_for_update: bool = False,
load_options: list[ExecutableOption] | None = None,
as_response: Literal[True],
schema: None = ...,
) -> Response[ModelType]: ...
@overload
@classmethod
async def get( # pragma: no cover
cls: type[Self],
session: AsyncSession,
filters: list[Any],
*,
joins: JoinType | None = None,
outer_join: bool = False,
with_for_update: bool = False,
load_options: list[ExecutableOption] | None = None,
as_response: Literal[False] = ...,
schema: None = ...,
) -> ModelType: ...
@@ -315,9 +272,8 @@ class AsyncCrud(Generic[ModelType]):
outer_join: bool = False,
with_for_update: bool = False,
load_options: list[ExecutableOption] | None = None,
as_response: bool = False,
schema: type[BaseModel] | None = None,
) -> ModelType | Response[ModelType] | Response[Any]:
) -> ModelType | Response[Any]:
"""Get exactly one record. Raises NotFoundError if not found.
Args:
@@ -327,25 +283,16 @@ class AsyncCrud(Generic[ModelType]):
outer_join: Use LEFT OUTER JOIN instead of INNER JOIN
with_for_update: Lock the row for update
load_options: SQLAlchemy loader options (e.g., selectinload)
as_response: Deprecated. Use ``schema`` instead. Will be removed in v2.0.
schema: Pydantic schema to serialize the result into. When provided,
the result is automatically wrapped in a ``Response[schema]``.
Returns:
Model instance, or ``Response[schema]`` when ``schema`` is given,
or ``Response[ModelType]`` when ``as_response=True`` (deprecated).
Model instance, or ``Response[schema]`` when ``schema`` is given.
Raises:
NotFoundError: If no record found
MultipleResultsFound: If more than one record found
"""
if as_response and schema is None:
warnings.warn(
"as_response is deprecated and will be removed in v2.0. "
"Use schema=YourSchema instead.",
DeprecationWarning,
stacklevel=2,
)
q = select(cls.model)
if joins:
for model, condition in joins:
@@ -364,7 +311,7 @@ class AsyncCrud(Generic[ModelType]):
if not item:
raise NotFoundError()
result = cast(ModelType, item)
if as_response or schema:
if schema:
data_out = schema.model_validate(result) if schema else result
return Response(data=data_out)
return result
@@ -466,10 +413,8 @@ class AsyncCrud(Generic[ModelType]):
exclude_unset: bool = True,
exclude_none: bool = False,
schema: type[SchemaType],
as_response: bool = ...,
) -> Response[SchemaType]: ...
# Backward-compatible - will be removed in v2.0
@overload
@classmethod
async def update( # pragma: no cover
@@ -480,21 +425,6 @@ class AsyncCrud(Generic[ModelType]):
*,
exclude_unset: bool = True,
exclude_none: bool = False,
as_response: Literal[True],
schema: None = ...,
) -> Response[ModelType]: ...
@overload
@classmethod
async def update( # pragma: no cover
cls: type[Self],
session: AsyncSession,
obj: BaseModel,
filters: list[Any],
*,
exclude_unset: bool = True,
exclude_none: bool = False,
as_response: Literal[False] = ...,
schema: None = ...,
) -> ModelType: ...
@@ -507,9 +437,8 @@ class AsyncCrud(Generic[ModelType]):
*,
exclude_unset: bool = True,
exclude_none: bool = False,
as_response: bool = False,
schema: type[BaseModel] | None = None,
) -> ModelType | Response[ModelType] | Response[Any]:
) -> ModelType | Response[Any]:
"""Update a record in the database.
Args:
@@ -518,24 +447,15 @@ class AsyncCrud(Generic[ModelType]):
filters: List of SQLAlchemy filter conditions
exclude_unset: Exclude fields not explicitly set in the schema
exclude_none: Exclude fields with None value
as_response: Deprecated. Use ``schema`` instead. Will be removed in v2.0.
schema: Pydantic schema to serialize the result into. When provided,
the result is automatically wrapped in a ``Response[schema]``.
Returns:
Updated model instance, or ``Response[schema]`` when ``schema`` is given,
or ``Response[ModelType]`` when ``as_response=True`` (deprecated).
Updated model instance, or ``Response[schema]`` when ``schema`` is given.
Raises:
NotFoundError: If no record found
"""
if as_response and schema is None:
warnings.warn(
"as_response is deprecated and will be removed in v2.0. "
"Use schema=YourSchema instead.",
DeprecationWarning,
stacklevel=2,
)
async with get_transaction(session):
m2m_exclude = cls._m2m_schema_fields()
@@ -565,7 +485,7 @@ class AsyncCrud(Generic[ModelType]):
for rel_attr, related_instances in m2m_resolved.items():
setattr(db_model, rel_attr, related_instances)
await session.refresh(db_model)
if as_response or schema:
if schema:
data_out = schema.model_validate(db_model) if schema else db_model
return Response(data=data_out)
return db_model
@@ -623,7 +543,7 @@ class AsyncCrud(Generic[ModelType]):
session: AsyncSession,
filters: list[Any],
*,
as_response: Literal[True],
return_response: Literal[True],
) -> Response[None]: ...
@overload
@@ -633,8 +553,8 @@ class AsyncCrud(Generic[ModelType]):
session: AsyncSession,
filters: list[Any],
*,
as_response: Literal[False] = ...,
) -> bool: ...
return_response: Literal[False] = ...,
) -> None: ...
@classmethod
async def delete(
@@ -642,33 +562,26 @@ class AsyncCrud(Generic[ModelType]):
session: AsyncSession,
filters: list[Any],
*,
as_response: bool = False,
) -> bool | Response[None]:
return_response: bool = False,
) -> None | Response[None]:
"""Delete records from the database.
Args:
session: DB async session
filters: List of SQLAlchemy filter conditions
as_response: Deprecated. Will be removed in v2.0. When ``True``,
returns ``Response[None]`` instead of ``bool``.
return_response: When ``True``, returns ``Response[None]`` instead
of ``None``. Useful for API endpoints that expect a consistent
response envelope.
Returns:
``True`` if deletion was executed, or ``Response[None]`` when
``as_response=True`` (deprecated).
``None``, or ``Response[None]`` when ``return_response=True``.
"""
if as_response:
warnings.warn(
"as_response is deprecated and will be removed in v2.0. "
"Use schema=YourSchema instead.",
DeprecationWarning,
stacklevel=2,
)
async with get_transaction(session):
q = sql_delete(cls.model).where(and_(*filters))
await session.execute(q)
if as_response:
if return_response:
return Response(data=None)
return True
return None
@classmethod
async def count(
@@ -735,47 +648,6 @@ class AsyncCrud(Generic[ModelType]):
result = await session.execute(q)
return bool(result.scalar())
@overload
@classmethod
async def offset_paginate( # pragma: no cover
cls: type[Self],
session: AsyncSession,
*,
filters: list[Any] | None = None,
joins: JoinType | None = None,
outer_join: bool = False,
load_options: list[ExecutableOption] | None = None,
order_by: Any | None = None,
page: int = 1,
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[SchemaType],
) -> PaginatedResponse[SchemaType]: ...
# Backward-compatible - will be removed in v2.0
@overload
@classmethod
async def offset_paginate( # pragma: no cover
cls: type[Self],
session: AsyncSession,
*,
filters: list[Any] | None = None,
joins: JoinType | None = None,
outer_join: bool = False,
load_options: list[ExecutableOption] | None = None,
order_by: Any | None = None,
page: int = 1,
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: None = ...,
) -> PaginatedResponse[ModelType]: ...
@classmethod
async def offset_paginate(
cls: type[Self],
@@ -792,8 +664,8 @@ class AsyncCrud(Generic[ModelType]):
search_fields: Sequence[SearchFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel] | None = None,
) -> PaginatedResponse[ModelType] | PaginatedResponse[Any]:
schema: type[BaseModel],
) -> PaginatedResponse[Any]:
"""Get paginated results using offset-based pagination.
Args:
@@ -811,7 +683,7 @@ class AsyncCrud(Generic[ModelType]):
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 InvalidFacetFilterError for unknown keys.
schema: Optional Pydantic schema to serialize each item into.
schema: Pydantic schema to serialize each item into.
Returns:
PaginatedResponse with OffsetPagination metadata
@@ -870,9 +742,7 @@ class AsyncCrud(Generic[ModelType]):
q = q.offset(offset).limit(items_per_page)
result = await session.execute(q)
raw_items = cast(list[ModelType], result.unique().scalars().all())
items: list[Any] = (
[schema.model_validate(item) for item in raw_items] if schema else raw_items
)
items: list[Any] = [schema.model_validate(item) for item in raw_items]
# Count query (with same joins and filters)
pk_col = cls.model.__mapper__.primary_key[0]
@@ -923,50 +793,6 @@ class AsyncCrud(Generic[ModelType]):
filter_attributes=filter_attributes,
)
# Backward-compatible - will be removed in v2.0
paginate = offset_paginate
@overload
@classmethod
async def cursor_paginate( # pragma: no cover
cls: type[Self],
session: AsyncSession,
*,
cursor: str | None = None,
filters: list[Any] | None = None,
joins: JoinType | None = None,
outer_join: bool = False,
load_options: list[ExecutableOption] | None = None,
order_by: Any | 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[SchemaType],
) -> PaginatedResponse[SchemaType]: ...
# Backward-compatible - will be removed in v2.0
@overload
@classmethod
async def cursor_paginate( # pragma: no cover
cls: type[Self],
session: AsyncSession,
*,
cursor: str | None = None,
filters: list[Any] | None = None,
joins: JoinType | None = None,
outer_join: bool = False,
load_options: list[ExecutableOption] | None = None,
order_by: Any | 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: None = ...,
) -> PaginatedResponse[ModelType]: ...
@classmethod
async def cursor_paginate(
cls: type[Self],
@@ -983,8 +809,8 @@ class AsyncCrud(Generic[ModelType]):
search_fields: Sequence[SearchFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel] | None = None,
) -> PaginatedResponse[ModelType] | PaginatedResponse[Any]:
schema: type[BaseModel],
) -> PaginatedResponse[Any]:
"""Get paginated results using cursor-based pagination.
Args:
@@ -1110,11 +936,7 @@ class AsyncCrud(Generic[ModelType]):
if cursor is not None and items_page:
prev_cursor = _encode_cursor(getattr(items_page[0], cursor_col_name))
items: list[Any] = (
[schema.model_validate(item) for item in items_page]
if schema
else items_page
)
items: list[Any] = [schema.model_validate(item) for item in items_page]
# Build facets
resolved_facet_fields = (

View File

@@ -10,7 +10,6 @@ __all__ = [
"CursorPagination",
"ErrorResponse",
"OffsetPagination",
"Pagination",
"PaginatedResponse",
"PydanticBase",
"Response",
@@ -108,10 +107,6 @@ class OffsetPagination(PydanticBase):
has_more: bool
# Backward-compatible - will be removed in v2.0
Pagination = OffsetPagination
class CursorPagination(PydanticBase):
"""Pagination metadata for cursor-based list responses.

View File

@@ -162,6 +162,7 @@ class UserRead(PydanticBase):
id: uuid.UUID
username: str
is_active: bool = True
class UserUpdate(BaseModel):
@@ -218,12 +219,26 @@ class PostM2MUpdate(BaseModel):
tag_ids: list[uuid.UUID] | None = None
class IntRoleRead(PydanticBase):
"""Schema for reading an IntRole."""
id: int
name: str
class IntRoleCreate(BaseModel):
"""Schema for creating an IntRole."""
name: str
class EventRead(PydanticBase):
"""Schema for reading an Event."""
id: uuid.UUID
name: str
class EventCreate(BaseModel):
"""Schema for creating an Event."""
@@ -232,6 +247,13 @@ class EventCreate(BaseModel):
scheduled_date: datetime.date
class ProductRead(PydanticBase):
"""Schema for reading a Product."""
id: uuid.UUID
name: str
class ProductCreate(BaseModel):
"""Schema for creating a Product."""

View File

@@ -15,8 +15,10 @@ from .conftest import (
EventCrud,
EventDateCursorCrud,
EventDateTimeCursorCrud,
EventRead,
IntRoleCreate,
IntRoleCursorCrud,
IntRoleRead,
Post,
PostCreate,
PostCrud,
@@ -26,6 +28,7 @@ from .conftest import (
ProductCreate,
ProductCrud,
ProductNumericCursorCrud,
ProductRead,
Role,
RoleCreate,
RoleCrud,
@@ -169,7 +172,14 @@ class TestDefaultLoadOptionsIntegration:
async def test_default_load_options_applied_to_paginate(
self, db_session: AsyncSession
):
"""default_load_options loads relationships automatically on paginate()."""
"""default_load_options loads relationships automatically on offset_paginate()."""
from fastapi_toolsets.schemas import PydanticBase
class UserWithRoleRead(PydanticBase):
id: uuid.UUID
username: str
role: RoleRead | None = None
UserWithDefaultLoad = CrudFactory(
User, default_load_options=[selectinload(User.role)]
)
@@ -178,7 +188,9 @@ class TestDefaultLoadOptionsIntegration:
db_session,
UserCreate(username="alice", email="alice@test.com", role_id=role.id),
)
result = await UserWithDefaultLoad.paginate(db_session)
result = await UserWithDefaultLoad.offset_paginate(
db_session, schema=UserWithRoleRead
)
assert result.data[0].role is not None
assert result.data[0].role.name == "admin"
@@ -430,7 +442,7 @@ class TestCrudDelete:
role = await RoleCrud.create(db_session, RoleCreate(name="to_delete"))
result = await RoleCrud.delete(db_session, [Role.id == role.id])
assert result is True
assert result is None
assert await RoleCrud.first(db_session, [Role.id == role.id]) is None
@pytest.mark.anyio
@@ -454,6 +466,20 @@ class TestCrudDelete:
assert len(remaining) == 1
assert remaining[0].username == "u3"
@pytest.mark.anyio
async def test_delete_return_response(self, db_session: AsyncSession):
"""Delete with return_response=True returns Response[None]."""
from fastapi_toolsets.schemas import Response
role = await RoleCrud.create(db_session, RoleCreate(name="to_delete_resp"))
result = await RoleCrud.delete(
db_session, [Role.id == role.id], return_response=True
)
assert isinstance(result, Response)
assert result.data is None
assert await RoleCrud.first(db_session, [Role.id == role.id]) is None
class TestCrudExists:
"""Tests for CRUD exists operations."""
@@ -594,7 +620,9 @@ class TestCrudPaginate:
from fastapi_toolsets.schemas import OffsetPagination
result = await RoleCrud.paginate(db_session, page=1, items_per_page=10)
result = await RoleCrud.offset_paginate(
db_session, page=1, items_per_page=10, schema=RoleRead
)
assert isinstance(result.pagination, OffsetPagination)
assert len(result.data) == 10
@@ -609,7 +637,9 @@ class TestCrudPaginate:
for i in range(25):
await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
result = await RoleCrud.paginate(db_session, page=3, items_per_page=10)
result = await RoleCrud.offset_paginate(
db_session, page=3, items_per_page=10, schema=RoleRead
)
assert len(result.data) == 5
assert result.pagination.has_more is False
@@ -629,11 +659,12 @@ class TestCrudPaginate:
from fastapi_toolsets.schemas import OffsetPagination
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
filters=[User.is_active == True], # noqa: E712
page=1,
items_per_page=10,
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -646,11 +677,12 @@ class TestCrudPaginate:
await RoleCrud.create(db_session, RoleCreate(name="alpha"))
await RoleCrud.create(db_session, RoleCreate(name="bravo"))
result = await RoleCrud.paginate(
result = await RoleCrud.offset_paginate(
db_session,
order_by=Role.name,
page=1,
items_per_page=10,
schema=RoleRead,
)
names = [r.name for r in result.data]
@@ -855,12 +887,13 @@ class TestCrudJoins:
from fastapi_toolsets.schemas import OffsetPagination
# Paginate users with published posts
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
joins=[(Post, Post.author_id == User.id)],
filters=[Post.is_published == True], # noqa: E712
page=1,
items_per_page=10,
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -889,12 +922,13 @@ class TestCrudJoins:
from fastapi_toolsets.schemas import OffsetPagination
# Paginate with outer join
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
joins=[(Post, Post.author_id == User.id)],
outer_join=True,
page=1,
items_per_page=10,
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -931,70 +965,6 @@ class TestCrudJoins:
assert users[0].username == "multi_join"
class TestAsResponse:
"""Tests for as_response parameter (deprecated, kept for backward compat)."""
@pytest.mark.anyio
async def test_create_as_response(self, db_session: AsyncSession):
"""Create with as_response=True returns Response and emits DeprecationWarning."""
from fastapi_toolsets.schemas import Response
data = RoleCreate(name="response_role")
with pytest.warns(DeprecationWarning, match="as_response is deprecated"):
result = await RoleCrud.create(db_session, data, as_response=True)
assert isinstance(result, Response)
assert result.data is not None
assert result.data.name == "response_role"
@pytest.mark.anyio
async def test_get_as_response(self, db_session: AsyncSession):
"""Get with as_response=True returns Response and emits DeprecationWarning."""
from fastapi_toolsets.schemas import Response
created = await RoleCrud.create(db_session, RoleCreate(name="get_response"))
with pytest.warns(DeprecationWarning, match="as_response is deprecated"):
result = await RoleCrud.get(
db_session, [Role.id == created.id], as_response=True
)
assert isinstance(result, Response)
assert result.data is not None
assert result.data.id == created.id
@pytest.mark.anyio
async def test_update_as_response(self, db_session: AsyncSession):
"""Update with as_response=True returns Response and emits DeprecationWarning."""
from fastapi_toolsets.schemas import Response
created = await RoleCrud.create(db_session, RoleCreate(name="old_name"))
with pytest.warns(DeprecationWarning, match="as_response is deprecated"):
result = await RoleCrud.update(
db_session,
RoleUpdate(name="new_name"),
[Role.id == created.id],
as_response=True,
)
assert isinstance(result, Response)
assert result.data is not None
assert result.data.name == "new_name"
@pytest.mark.anyio
async def test_delete_as_response(self, db_session: AsyncSession):
"""Delete with as_response=True returns Response and emits DeprecationWarning."""
from fastapi_toolsets.schemas import Response
created = await RoleCrud.create(db_session, RoleCreate(name="to_delete"))
with pytest.warns(DeprecationWarning, match="as_response is deprecated"):
result = await RoleCrud.delete(
db_session, [Role.id == created.id], as_response=True
)
assert isinstance(result, Response)
assert result.data is None
class TestCrudFactoryM2M:
"""Tests for CrudFactory with m2m_fields parameter."""
@@ -1475,92 +1445,35 @@ class TestSchemaResponse:
assert isinstance(result, Response)
@pytest.mark.anyio
async def test_paginate_with_schema(self, db_session: AsyncSession):
"""paginate with schema returns PaginatedResponse[SchemaType]."""
async def test_offset_paginate_with_schema(self, db_session: AsyncSession):
"""offset_paginate with schema returns PaginatedResponse[SchemaType]."""
from fastapi_toolsets.schemas import PaginatedResponse
await RoleCrud.create(db_session, RoleCreate(name="p_role1"))
await RoleCrud.create(db_session, RoleCreate(name="p_role2"))
result = await RoleCrud.paginate(db_session, schema=RoleRead)
result = await RoleCrud.offset_paginate(db_session, schema=RoleRead)
assert isinstance(result, PaginatedResponse)
assert len(result.data) == 2
assert all(isinstance(item, RoleRead) for item in result.data)
@pytest.mark.anyio
async def test_paginate_schema_filters_fields(self, db_session: AsyncSession):
"""paginate with schema only exposes schema fields per item."""
async def test_offset_paginate_schema_filters_fields(
self, db_session: AsyncSession
):
"""offset_paginate with schema only exposes schema fields per item."""
await UserCrud.create(
db_session,
UserCreate(username="pg_user", email="pg@test.com"),
)
result = await UserCrud.paginate(db_session, schema=UserRead)
result = await UserCrud.offset_paginate(db_session, schema=UserRead)
assert isinstance(result.data[0], UserRead)
assert result.data[0].username == "pg_user"
assert not hasattr(result.data[0], "email")
@pytest.mark.anyio
async def test_as_response_true_without_schema_unchanged(
self, db_session: AsyncSession
):
"""as_response=True without schema still returns Response[ModelType] with a warning."""
from fastapi_toolsets.schemas import Response
created = await RoleCrud.create(db_session, RoleCreate(name="compat"))
with pytest.warns(DeprecationWarning, match="as_response is deprecated"):
result = await RoleCrud.get(
db_session, [Role.id == created.id], as_response=True
)
assert isinstance(result, Response)
assert isinstance(result.data, Role)
@pytest.mark.anyio
async def test_schema_with_explicit_as_response_true(
self, db_session: AsyncSession
):
"""schema combined with explicit as_response=True works correctly."""
from fastapi_toolsets.schemas import Response
created = await RoleCrud.create(db_session, RoleCreate(name="combined"))
result = await RoleCrud.get(
db_session,
[Role.id == created.id],
as_response=True,
schema=RoleRead,
)
assert isinstance(result, Response)
assert isinstance(result.data, RoleRead)
class TestPaginateAlias:
"""Tests that paginate is a backward-compatible alias for offset_paginate."""
def test_paginate_is_alias_of_offset_paginate(self):
"""paginate and offset_paginate are the same underlying function."""
assert RoleCrud.paginate.__func__ is RoleCrud.offset_paginate.__func__
@pytest.mark.anyio
async def test_paginate_alias_returns_offset_pagination(
self, db_session: AsyncSession
):
"""paginate() still works and returns PaginatedResponse with OffsetPagination."""
from fastapi_toolsets.schemas import OffsetPagination, PaginatedResponse
for i in range(3):
await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
result = await RoleCrud.paginate(db_session, page=1, items_per_page=10)
assert isinstance(result, PaginatedResponse)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 3
assert result.pagination.page == 1
class TestCursorPaginate:
"""Tests for cursor-based pagination via cursor_paginate()."""
@@ -1573,7 +1486,9 @@ class TestCursorPaginate:
for i in range(25):
await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
result = await RoleCursorCrud.cursor_paginate(db_session, items_per_page=10)
result = await RoleCursorCrud.cursor_paginate(
db_session, items_per_page=10, schema=RoleRead
)
assert isinstance(result, PaginatedResponse)
assert isinstance(result.pagination, CursorPagination)
@@ -1591,7 +1506,9 @@ class TestCursorPaginate:
for i in range(5):
await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
result = await RoleCursorCrud.cursor_paginate(db_session, items_per_page=10)
result = await RoleCursorCrud.cursor_paginate(
db_session, items_per_page=10, schema=RoleRead
)
assert isinstance(result.pagination, CursorPagination)
assert len(result.data) == 5
@@ -1606,14 +1523,16 @@ class TestCursorPaginate:
from fastapi_toolsets.schemas import CursorPagination
page1 = await RoleCursorCrud.cursor_paginate(db_session, items_per_page=10)
page1 = await RoleCursorCrud.cursor_paginate(
db_session, items_per_page=10, schema=RoleRead
)
assert isinstance(page1.pagination, CursorPagination)
assert len(page1.data) == 10
assert page1.pagination.has_more is True
cursor = page1.pagination.next_cursor
page2 = await RoleCursorCrud.cursor_paginate(
db_session, cursor=cursor, items_per_page=10
db_session, cursor=cursor, items_per_page=10, schema=RoleRead
)
assert isinstance(page2.pagination, CursorPagination)
assert len(page2.data) == 5
@@ -1628,12 +1547,15 @@ class TestCursorPaginate:
from fastapi_toolsets.schemas import CursorPagination
page1 = await RoleCursorCrud.cursor_paginate(db_session, items_per_page=4)
page1 = await RoleCursorCrud.cursor_paginate(
db_session, items_per_page=4, schema=RoleRead
)
assert isinstance(page1.pagination, CursorPagination)
page2 = await RoleCursorCrud.cursor_paginate(
db_session,
cursor=page1.pagination.next_cursor,
items_per_page=4,
schema=RoleRead,
)
ids_page1 = {r.id for r in page1.data}
@@ -1646,7 +1568,9 @@ class TestCursorPaginate:
"""cursor_paginate on an empty table returns empty data with no cursor."""
from fastapi_toolsets.schemas import CursorPagination
result = await RoleCursorCrud.cursor_paginate(db_session, items_per_page=10)
result = await RoleCursorCrud.cursor_paginate(
db_session, items_per_page=10, schema=RoleRead
)
assert isinstance(result.pagination, CursorPagination)
assert result.data == []
@@ -1671,6 +1595,7 @@ class TestCursorPaginate:
db_session,
filters=[User.is_active == True], # noqa: E712
items_per_page=20,
schema=UserRead,
)
assert len(result.data) == 5
@@ -1703,7 +1628,9 @@ class TestCursorPaginate:
for i in range(5):
await RoleNameCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
result = await RoleNameCrud.cursor_paginate(db_session, items_per_page=3)
result = await RoleNameCrud.cursor_paginate(
db_session, items_per_page=3, schema=RoleRead
)
assert isinstance(result.pagination, CursorPagination)
assert len(result.data) == 3
@@ -1714,7 +1641,7 @@ class TestCursorPaginate:
async def test_raises_without_cursor_column(self, db_session: AsyncSession):
"""cursor_paginate raises ValueError when cursor_column is not configured."""
with pytest.raises(ValueError, match="cursor_column is not set"):
await RoleCrud.cursor_paginate(db_session)
await RoleCrud.cursor_paginate(db_session, schema=RoleRead)
class TestCursorPaginatePrevCursor:
@@ -1728,7 +1655,9 @@ class TestCursorPaginatePrevCursor:
from fastapi_toolsets.schemas import CursorPagination
result = await RoleCursorCrud.cursor_paginate(db_session, items_per_page=3)
result = await RoleCursorCrud.cursor_paginate(
db_session, items_per_page=3, schema=RoleRead
)
assert isinstance(result.pagination, CursorPagination)
assert result.pagination.prev_cursor is None
@@ -1741,12 +1670,15 @@ class TestCursorPaginatePrevCursor:
from fastapi_toolsets.schemas import CursorPagination
page1 = await RoleCursorCrud.cursor_paginate(db_session, items_per_page=5)
page1 = await RoleCursorCrud.cursor_paginate(
db_session, items_per_page=5, schema=RoleRead
)
assert isinstance(page1.pagination, CursorPagination)
page2 = await RoleCursorCrud.cursor_paginate(
db_session,
cursor=page1.pagination.next_cursor,
items_per_page=5,
schema=RoleRead,
)
assert isinstance(page2.pagination, CursorPagination)
assert page2.pagination.prev_cursor is not None
@@ -1762,12 +1694,15 @@ class TestCursorPaginatePrevCursor:
from fastapi_toolsets.schemas import CursorPagination
page1 = await RoleCursorCrud.cursor_paginate(db_session, items_per_page=5)
page1 = await RoleCursorCrud.cursor_paginate(
db_session, items_per_page=5, schema=RoleRead
)
assert isinstance(page1.pagination, CursorPagination)
page2 = await RoleCursorCrud.cursor_paginate(
db_session,
cursor=page1.pagination.next_cursor,
items_per_page=5,
schema=RoleRead,
)
assert isinstance(page2.pagination, CursorPagination)
assert page2.pagination.prev_cursor is not None
@@ -1802,6 +1737,7 @@ class TestCursorPaginateWithSearch:
db_session,
search="admin",
items_per_page=20,
schema=RoleRead,
)
assert len(result.data) == 5
@@ -1836,6 +1772,7 @@ class TestCursorPaginateExtraOptions:
db_session,
joins=[(Role, User.role_id == Role.id)],
items_per_page=20,
schema=UserRead,
)
assert isinstance(result.pagination, CursorPagination)
@@ -1867,6 +1804,7 @@ class TestCursorPaginateExtraOptions:
joins=[(Role, User.role_id == Role.id)],
outer_join=True,
items_per_page=20,
schema=UserRead,
)
assert isinstance(result.pagination, CursorPagination)
@@ -1876,7 +1814,12 @@ class TestCursorPaginateExtraOptions:
@pytest.mark.anyio
async def test_with_load_options(self, db_session: AsyncSession):
"""cursor_paginate passes load_options to the query."""
from fastapi_toolsets.schemas import CursorPagination
from fastapi_toolsets.schemas import CursorPagination, PydanticBase
class UserWithRoleRead(PydanticBase):
id: uuid.UUID
username: str
role: RoleRead | None = None
role = await RoleCrud.create(db_session, RoleCreate(name="manager"))
for i in range(3):
@@ -1893,6 +1836,7 @@ class TestCursorPaginateExtraOptions:
db_session,
load_options=[selectinload(User.role)],
items_per_page=20,
schema=UserWithRoleRead,
)
assert isinstance(result.pagination, CursorPagination)
@@ -1912,6 +1856,7 @@ class TestCursorPaginateExtraOptions:
db_session,
order_by=Role.name.desc(),
items_per_page=3,
schema=RoleRead,
)
assert isinstance(result.pagination, CursorPagination)
@@ -1925,7 +1870,9 @@ class TestCursorPaginateExtraOptions:
for i in range(5):
await IntRoleCursorCrud.create(db_session, IntRoleCreate(name=f"role{i}"))
page1 = await IntRoleCursorCrud.cursor_paginate(db_session, items_per_page=3)
page1 = await IntRoleCursorCrud.cursor_paginate(
db_session, items_per_page=3, schema=IntRoleRead
)
assert isinstance(page1.pagination, CursorPagination)
assert len(page1.data) == 3
@@ -1935,6 +1882,7 @@ class TestCursorPaginateExtraOptions:
db_session,
cursor=page1.pagination.next_cursor,
items_per_page=3,
schema=IntRoleRead,
)
assert isinstance(page2.pagination, CursorPagination)
@@ -1955,7 +1903,9 @@ class TestCursorPaginateExtraOptions:
await RoleCrud.create(db_session, RoleCreate(name="role01"))
# First page succeeds (no cursor to decode)
page1 = await RoleNameCursorCrud.cursor_paginate(db_session, items_per_page=1)
page1 = await RoleNameCursorCrud.cursor_paginate(
db_session, items_per_page=1, schema=RoleRead
)
assert page1.pagination.has_more is True
assert isinstance(page1.pagination, CursorPagination)
@@ -1965,6 +1915,7 @@ class TestCursorPaginateExtraOptions:
db_session,
cursor=page1.pagination.next_cursor,
items_per_page=1,
schema=RoleRead,
)
@@ -2003,6 +1954,7 @@ class TestCursorPaginateSearchJoins:
search="administrator",
search_fields=[(User.role, Role.name)],
items_per_page=20,
schema=UserRead,
)
assert isinstance(result.pagination, CursorPagination)
@@ -2049,7 +2001,7 @@ class TestCursorPaginateColumnTypes:
)
page1 = await EventDateTimeCursorCrud.cursor_paginate(
db_session, items_per_page=3
db_session, items_per_page=3, schema=EventRead
)
assert isinstance(page1.pagination, CursorPagination)
@@ -2060,6 +2012,7 @@ class TestCursorPaginateColumnTypes:
db_session,
cursor=page1.pagination.next_cursor,
items_per_page=3,
schema=EventRead,
)
assert isinstance(page2.pagination, CursorPagination)
@@ -2087,7 +2040,9 @@ class TestCursorPaginateColumnTypes:
),
)
page1 = await EventDateCursorCrud.cursor_paginate(db_session, items_per_page=3)
page1 = await EventDateCursorCrud.cursor_paginate(
db_session, items_per_page=3, schema=EventRead
)
assert isinstance(page1.pagination, CursorPagination)
assert len(page1.data) == 3
@@ -2097,6 +2052,7 @@ class TestCursorPaginateColumnTypes:
db_session,
cursor=page1.pagination.next_cursor,
items_per_page=3,
schema=EventRead,
)
assert isinstance(page2.pagination, CursorPagination)
@@ -2123,7 +2079,7 @@ class TestCursorPaginateColumnTypes:
)
page1 = await ProductNumericCursorCrud.cursor_paginate(
db_session, items_per_page=3
db_session, items_per_page=3, schema=ProductRead
)
assert isinstance(page1.pagination, CursorPagination)
@@ -2134,6 +2090,7 @@ class TestCursorPaginateColumnTypes:
db_session,
cursor=page1.pagination.next_cursor,
items_per_page=3,
schema=ProductRead,
)
assert isinstance(page2.pagination, CursorPagination)

View File

@@ -20,6 +20,7 @@ from .conftest import (
User,
UserCreate,
UserCrud,
UserRead,
)
@@ -39,10 +40,11 @@ class TestPaginateSearch:
db_session, UserCreate(username="bob_smith", email="bob@test.com")
)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="doe",
search_fields=[User.username],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -58,10 +60,11 @@ class TestPaginateSearch:
db_session, UserCreate(username="company_bob", email="bob@other.com")
)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="company",
search_fields=[User.username, User.email],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -86,10 +89,11 @@ class TestPaginateSearch:
UserCreate(username="user1", email="u1@test.com", role_id=user_role.id),
)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="admin",
search_fields=[(User.role, Role.name)],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -105,10 +109,11 @@ class TestPaginateSearch:
)
# Search "admin" in username OR role.name
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="admin",
search_fields=[User.username, (User.role, Role.name)],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -121,10 +126,11 @@ class TestPaginateSearch:
db_session, UserCreate(username="JohnDoe", email="j@test.com")
)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="johndoe",
search_fields=[User.username],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -138,19 +144,21 @@ class TestPaginateSearch:
)
# Should not find (case mismatch)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search=SearchConfig(query="johndoe", case_sensitive=True),
search_fields=[User.username],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 0
# Should find (case match)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search=SearchConfig(query="JohnDoe", case_sensitive=True),
search_fields=[User.username],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 1
@@ -165,11 +173,13 @@ class TestPaginateSearch:
db_session, UserCreate(username="user2", email="u2@test.com")
)
result = await UserCrud.paginate(db_session, search="")
result = await UserCrud.offset_paginate(db_session, search="", schema=UserRead)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 2
result = await UserCrud.paginate(db_session, search=None)
result = await UserCrud.offset_paginate(
db_session, search=None, schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 2
@@ -185,11 +195,12 @@ class TestPaginateSearch:
UserCreate(username="inactive_john", email="ij@test.com", is_active=False),
)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
filters=[User.is_active == True], # noqa: E712
search="john",
search_fields=[User.username],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -203,7 +214,9 @@ class TestPaginateSearch:
db_session, UserCreate(username="findme", email="other@test.com")
)
result = await UserCrud.paginate(db_session, search="findme")
result = await UserCrud.offset_paginate(
db_session, search="findme", schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 1
@@ -215,10 +228,11 @@ class TestPaginateSearch:
db_session, UserCreate(username="john", email="j@test.com")
)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="nonexistent",
search_fields=[User.username],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -234,12 +248,13 @@ class TestPaginateSearch:
UserCreate(username=f"user_{i}", email=f"user{i}@test.com"),
)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="user_",
search_fields=[User.username],
page=1,
items_per_page=5,
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -261,10 +276,11 @@ class TestPaginateSearch:
)
# Search in username, not in role
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="role",
search_fields=[User.username],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -283,11 +299,12 @@ class TestPaginateSearch:
db_session, UserCreate(username="bob", email="b@test.com")
)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="@test.com",
search_fields=[User.email],
order_by=User.username,
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -307,10 +324,11 @@ class TestPaginateSearch:
)
# Search by UUID (partial match)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search="12345678",
search_fields=[User.id, User.username],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -360,10 +378,11 @@ class TestSearchConfig:
)
# 'john' must be in username AND email
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search=SearchConfig(query="john", match_mode="all"),
search_fields=[User.username, User.email],
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -377,9 +396,10 @@ class TestSearchConfig:
db_session, UserCreate(username="test", email="findme@test.com")
)
result = await UserCrud.paginate(
result = await UserCrud.offset_paginate(
db_session,
search=SearchConfig(query="findme", fields=[User.email]),
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -475,7 +495,7 @@ class TestFacetsNotSet:
db_session, UserCreate(username="alice", email="a@test.com")
)
result = await UserCrud.offset_paginate(db_session)
result = await UserCrud.offset_paginate(db_session, schema=UserRead)
assert result.filter_attributes is None
@@ -487,7 +507,7 @@ class TestFacetsNotSet:
db_session, UserCreate(username="alice", email="a@test.com")
)
result = await UserCursorCrud.cursor_paginate(db_session)
result = await UserCursorCrud.cursor_paginate(db_session, schema=UserRead)
assert result.filter_attributes is None
@@ -506,7 +526,7 @@ class TestFacetsDirectColumn:
db_session, UserCreate(username="bob", email="b@test.com")
)
result = await UserFacetCrud.offset_paginate(db_session)
result = await UserFacetCrud.offset_paginate(db_session, schema=UserRead)
assert result.filter_attributes is not None
# Distinct usernames, sorted
@@ -525,7 +545,7 @@ class TestFacetsDirectColumn:
db_session, UserCreate(username="bob", email="b@test.com")
)
result = await UserFacetCursorCrud.cursor_paginate(db_session)
result = await UserFacetCursorCrud.cursor_paginate(db_session, schema=UserRead)
assert result.filter_attributes is not None
assert set(result.filter_attributes["email"]) == {"a@test.com", "b@test.com"}
@@ -541,7 +561,7 @@ class TestFacetsDirectColumn:
db_session, UserCreate(username="bob", email="b@test.com")
)
result = await UserFacetCrud.offset_paginate(db_session)
result = await UserFacetCrud.offset_paginate(db_session, schema=UserRead)
assert result.filter_attributes is not None
assert "username" in result.filter_attributes
@@ -558,7 +578,7 @@ class TestFacetsDirectColumn:
# Override: ask for email instead of username
result = await UserFacetCrud.offset_paginate(
db_session, facet_fields=[User.email]
db_session, facet_fields=[User.email], schema=UserRead
)
assert result.filter_attributes is not None
@@ -584,6 +604,7 @@ class TestFacetsRespectFilters:
result = await UserFacetCrud.offset_paginate(
db_session,
filters=[User.is_active == True], # noqa: E712
schema=UserRead,
)
assert result.filter_attributes is not None
@@ -614,7 +635,7 @@ class TestFacetsRelationship:
db_session, UserCreate(username="charlie", email="c@test.com")
)
result = await UserRelFacetCrud.offset_paginate(db_session)
result = await UserRelFacetCrud.offset_paginate(db_session, schema=UserRead)
assert result.filter_attributes is not None
assert set(result.filter_attributes["name"]) == {"admin", "editor"}
@@ -629,7 +650,7 @@ class TestFacetsRelationship:
db_session, UserCreate(username="norole", email="n@test.com")
)
result = await UserRelFacetCrud.offset_paginate(db_session)
result = await UserRelFacetCrud.offset_paginate(db_session, schema=UserRead)
assert result.filter_attributes is not None
assert result.filter_attributes["name"] == []
@@ -653,7 +674,10 @@ class TestFacetsRelationship:
)
result = await UserSearchFacetCrud.offset_paginate(
db_session, search="admin", search_fields=[(User.role, Role.name)]
db_session,
search="admin",
search_fields=[(User.role, Role.name)],
schema=UserRead,
)
assert result.filter_attributes is not None
@@ -675,7 +699,7 @@ class TestFilterBy:
)
result = await UserFacetCrud.offset_paginate(
db_session, filter_by={"username": "alice"}
db_session, filter_by={"username": "alice"}, schema=UserRead
)
assert len(result.data) == 1
@@ -698,7 +722,7 @@ class TestFilterBy:
)
result = await UserFacetCrud.offset_paginate(
db_session, filter_by={"username": ["alice", "bob"]}
db_session, filter_by={"username": ["alice", "bob"]}, schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
@@ -723,7 +747,7 @@ class TestFilterBy:
)
result = await UserRelFacetCrud.offset_paginate(
db_session, filter_by={"name": "admin"}
db_session, filter_by={"name": "admin"}, schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
@@ -746,6 +770,7 @@ class TestFilterBy:
db_session,
filters=[User.is_active == True], # noqa: E712
filter_by={"username": ["alice", "alice2"]},
schema=UserRead,
)
# Only alice passes both: is_active=True AND username IN [alice, alice2]
@@ -760,7 +785,7 @@ class TestFilterBy:
with pytest.raises(InvalidFacetFilterError) as exc_info:
await UserFacetCrud.offset_paginate(
db_session, filter_by={"nonexistent": "value"}
db_session, filter_by={"nonexistent": "value"}, schema=UserRead
)
assert exc_info.value.key == "nonexistent"
@@ -792,6 +817,7 @@ class TestFilterBy:
result = await UserRoleFacetCrud.offset_paginate(
db_session,
filter_by={"name": "admin", "id": str(admin.id)},
schema=UserRead,
)
assert isinstance(result.pagination, OffsetPagination)
@@ -812,7 +838,7 @@ class TestFilterBy:
)
result = await UserFacetCursorCrud.cursor_paginate(
db_session, filter_by={"username": "alice"}
db_session, filter_by={"username": "alice"}, schema=UserRead
)
assert len(result.data) == 1
@@ -836,7 +862,7 @@ class TestFilterBy:
)
result = await UserFacetCrud.offset_paginate(
db_session, filter_by=UserFilter(username="alice")
db_session, filter_by=UserFilter(username="alice"), schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
@@ -862,7 +888,7 @@ class TestFilterBy:
)
result = await UserFacetCursorCrud.cursor_paginate(
db_session, filter_by=UserFilter(username="alice")
db_session, filter_by=UserFilter(username="alice"), schema=UserRead
)
assert len(result.data) == 1
@@ -971,7 +997,9 @@ class TestFilterParamsSchema:
dep = UserFacetCrud.filter_params()
f = await dep(username=["alice"])
result = await UserFacetCrud.offset_paginate(db_session, filter_by=f)
result = await UserFacetCrud.offset_paginate(
db_session, filter_by=f, schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 1
@@ -992,7 +1020,9 @@ class TestFilterParamsSchema:
dep = UserFacetCursorCrud.filter_params()
f = await dep(username=["alice"])
result = await UserFacetCursorCrud.cursor_paginate(db_session, filter_by=f)
result = await UserFacetCursorCrud.cursor_paginate(
db_session, filter_by=f, schema=UserRead
)
assert len(result.data) == 1
assert result.data[0].username == "alice"
@@ -1010,7 +1040,9 @@ class TestFilterParamsSchema:
dep = UserFacetCrud.filter_params()
f = await dep() # all fields None
result = await UserFacetCrud.offset_paginate(db_session, filter_by=f)
result = await UserFacetCrud.offset_paginate(
db_session, filter_by=f, schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 2

View File

@@ -0,0 +1,271 @@
"""Live test for the docs/examples/pagination-search.md example.
Spins up the exact FastAPI app described in the example (sourced from
docs_src/examples/pagination_search/) and exercises it through a real HTTP
client against a real PostgreSQL database.
"""
import datetime
import pytest
from fastapi import FastAPI
from httpx import ASGITransport, AsyncClient
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from docs_src.examples.pagination_search.db import get_db
from docs_src.examples.pagination_search.models import Article, Base, Category
from docs_src.examples.pagination_search.routes import router
from .conftest import DATABASE_URL
def build_app(session: AsyncSession) -> FastAPI:
app = FastAPI()
async def override_get_db():
yield session
app.dependency_overrides[get_db] = override_get_db
app.include_router(router)
return app
@pytest.fixture(scope="function")
async def ex_db_session():
"""Isolated session for the example models (separate tables from conftest)."""
engine = create_async_engine(DATABASE_URL, echo=False)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
session = session_factory()
try:
yield session
finally:
await session.close()
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await engine.dispose()
@pytest.fixture
async def client(ex_db_session: AsyncSession):
app = build_app(ex_db_session)
async with AsyncClient(
transport=ASGITransport(app=app), base_url="http://test"
) as ac:
yield ac
async def seed(session: AsyncSession):
"""Insert representative fixture data."""
python = Category(name="python")
backend = Category(name="backend")
session.add_all([python, backend])
await session.flush()
now = datetime.datetime(2024, 1, 1, tzinfo=datetime.timezone.utc)
session.add_all(
[
Article(
title="FastAPI tips",
body="Ten useful tips for FastAPI.",
status="published",
published=True,
category_id=python.id,
created_at=now,
),
Article(
title="SQLAlchemy async",
body="How to use async SQLAlchemy.",
status="published",
published=True,
category_id=backend.id,
created_at=now + datetime.timedelta(seconds=1),
),
Article(
title="Draft notes",
body="Work in progress.",
status="draft",
published=False,
category_id=None,
created_at=now + datetime.timedelta(seconds=2),
),
]
)
await session.commit()
class TestAppSessionDep:
@pytest.mark.anyio
async def test_get_db_yields_async_session(self):
"""get_db yields a real AsyncSession when called directly."""
from docs_src.examples.pagination_search.db import get_db
gen = get_db()
session = await gen.__anext__()
assert isinstance(session, AsyncSession)
await session.close()
class TestOffsetPagination:
@pytest.mark.anyio
async def test_returns_all_articles(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
resp = await client.get("/articles/offset")
assert resp.status_code == 200
body = resp.json()
assert body["pagination"]["total_count"] == 3
assert len(body["data"]) == 3
@pytest.mark.anyio
async def test_pagination_page_size(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
resp = await client.get("/articles/offset?items_per_page=2&page=1")
assert resp.status_code == 200
body = resp.json()
assert len(body["data"]) == 2
assert body["pagination"]["total_count"] == 3
assert body["pagination"]["has_more"] is True
@pytest.mark.anyio
async def test_full_text_search(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
resp = await client.get("/articles/offset?search=fastapi")
assert resp.status_code == 200
body = resp.json()
assert body["pagination"]["total_count"] == 1
assert body["data"][0]["title"] == "FastAPI tips"
@pytest.mark.anyio
async def test_search_traverses_relationship(
self, client: AsyncClient, ex_db_session
):
await seed(ex_db_session)
# "python" matches Category.name, not Article.title or body
resp = await client.get("/articles/offset?search=python")
assert resp.status_code == 200
body = resp.json()
assert body["pagination"]["total_count"] == 1
assert body["data"][0]["title"] == "FastAPI tips"
@pytest.mark.anyio
async def test_facet_filter_scalar(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
resp = await client.get("/articles/offset?status=published")
assert resp.status_code == 200
body = resp.json()
assert body["pagination"]["total_count"] == 2
assert all(a["status"] == "published" for a in body["data"])
@pytest.mark.anyio
async def test_facet_filter_multi_value(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
resp = await client.get("/articles/offset?status=published&status=draft")
assert resp.status_code == 200
body = resp.json()
assert body["pagination"]["total_count"] == 3
@pytest.mark.anyio
async def test_filter_attributes_in_response(
self, client: AsyncClient, ex_db_session
):
await seed(ex_db_session)
resp = await client.get("/articles/offset")
assert resp.status_code == 200
body = resp.json()
fa = body["filter_attributes"]
assert set(fa["status"]) == {"draft", "published"}
# "name" is unique across all facet fields — no prefix needed
assert set(fa["name"]) == {"backend", "python"}
@pytest.mark.anyio
async def test_filter_attributes_scoped_to_filter(
self, client: AsyncClient, ex_db_session
):
await seed(ex_db_session)
resp = await client.get("/articles/offset?status=published")
body = resp.json()
# draft is filtered out → should not appear in filter_attributes
assert "draft" not in body["filter_attributes"]["status"]
@pytest.mark.anyio
async def test_search_and_filter_combined(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
resp = await client.get("/articles/offset?search=async&status=published")
assert resp.status_code == 200
body = resp.json()
assert body["pagination"]["total_count"] == 1
assert body["data"][0]["title"] == "SQLAlchemy async"
class TestCursorPagination:
@pytest.mark.anyio
async def test_first_page(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
resp = await client.get("/articles/cursor?items_per_page=2")
assert resp.status_code == 200
body = resp.json()
assert len(body["data"]) == 2
assert body["pagination"]["has_more"] is True
assert body["pagination"]["next_cursor"] is not None
assert body["pagination"]["prev_cursor"] is None
@pytest.mark.anyio
async def test_second_page(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
first = await client.get("/articles/cursor?items_per_page=2")
next_cursor = first.json()["pagination"]["next_cursor"]
resp = await client.get(
f"/articles/cursor?items_per_page=2&cursor={next_cursor}"
)
assert resp.status_code == 200
body = resp.json()
assert len(body["data"]) == 1
assert body["pagination"]["has_more"] is False
@pytest.mark.anyio
async def test_facet_filter(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
resp = await client.get("/articles/cursor?status=draft")
assert resp.status_code == 200
body = resp.json()
assert len(body["data"]) == 1
assert body["data"][0]["status"] == "draft"
@pytest.mark.anyio
async def test_full_text_search(self, client: AsyncClient, ex_db_session):
await seed(ex_db_session)
resp = await client.get("/articles/cursor?search=sqlalchemy")
assert resp.status_code == 200
body = resp.json()
assert len(body["data"]) == 1
assert body["data"][0]["title"] == "SQLAlchemy async"

View File

@@ -9,7 +9,6 @@ from fastapi_toolsets.schemas import (
ErrorResponse,
OffsetPagination,
PaginatedResponse,
Pagination,
Response,
ResponseStatus,
)
@@ -199,20 +198,6 @@ class TestOffsetPagination:
assert data["page"] == 2
assert data["has_more"] is True
def test_pagination_alias_is_offset_pagination(self):
"""Pagination is a backward-compatible alias for OffsetPagination."""
assert Pagination is OffsetPagination
def test_pagination_alias_constructs_offset_pagination(self):
"""Code using Pagination(...) still works unchanged."""
pagination = Pagination(
total_count=10,
items_per_page=5,
page=2,
has_more=False,
)
assert isinstance(pagination, OffsetPagination)
class TestCursorPagination:
"""Tests for CursorPagination schema."""
@@ -276,7 +261,7 @@ class TestPaginatedResponse:
def test_create_paginated_response(self):
"""Create PaginatedResponse with data and pagination."""
pagination = Pagination(
pagination = OffsetPagination(
total_count=30,
items_per_page=10,
page=1,
@@ -294,7 +279,7 @@ class TestPaginatedResponse:
def test_with_custom_message(self):
"""PaginatedResponse with custom message."""
pagination = Pagination(
pagination = OffsetPagination(
total_count=5,
items_per_page=10,
page=1,
@@ -310,7 +295,7 @@ class TestPaginatedResponse:
def test_empty_data(self):
"""PaginatedResponse with empty data."""
pagination = Pagination(
pagination = OffsetPagination(
total_count=0,
items_per_page=10,
page=1,
@@ -332,7 +317,7 @@ class TestPaginatedResponse:
id: int
name: str
pagination = Pagination(
pagination = OffsetPagination(
total_count=1,
items_per_page=10,
page=1,
@@ -347,7 +332,7 @@ class TestPaginatedResponse:
def test_serialization(self):
"""PaginatedResponse serializes correctly."""
pagination = Pagination(
pagination = OffsetPagination(
total_count=100,
items_per_page=10,
page=5,
@@ -385,16 +370,6 @@ class TestPaginatedResponse:
)
assert isinstance(response.pagination, CursorPagination)
def test_pagination_alias_accepted(self):
"""Constructing PaginatedResponse with Pagination (alias) still works."""
response = PaginatedResponse(
data=[],
pagination=Pagination(
total_count=0, items_per_page=10, page=1, has_more=False
),
)
assert isinstance(response.pagination, OffsetPagination)
class TestFromAttributes:
"""Tests for from_attributes config (ORM mode)."""

2
uv.lock generated
View File

@@ -251,7 +251,7 @@ wheels = [
[[package]]
name = "fastapi-toolsets"
version = "1.2.0"
version = "1.2.1"
source = { editable = "." }
dependencies = [
{ name = "asyncpg" },

View File

@@ -1,265 +1,35 @@
# ============================================================================
#
# The configuration produced by default is meant to highlight the features
# that Zensical provides and to serve as a starting point for your own
# projects.
#
# ============================================================================
[project]
# The site_name is shown in the page header and the browser window title
#
# Read more: https://zensical.org/docs/setup/basics/#site_name
site_name = "FastAPI Toolsets"
# The site_description is included in the HTML head and should contain a
# meaningful description of the site content for use by search engines.
#
# Read more: https://zensical.org/docs/setup/basics/#site_description
site_description = "Production-ready utilities for FastAPI applications."
# The site_author attribute. This is used in the HTML head element.
#
# Read more: https://zensical.org/docs/setup/basics/#site_author
site_author = "d3vyce"
# The site_url is the canonical URL for your site. When building online
# documentation you should set this.
# Read more: https://zensical.org/docs/setup/basics/#site_url
site_url = "https://fastapi-toolsets.d3vyce.fr"
# The copyright notice appears in the page footer and can contain an HTML
# fragment.
#
# Read more: https://zensical.org/docs/setup/basics/#copyright
copyright = """
Copyright &copy; 2026 d3vyce
"""
copyright = "Copyright &copy; 2026 d3vyce"
repo_url = "https://github.com/d3vyce/fastapi-toolsets"
# Zensical supports both implicit navigation and explicitly defined navigation.
# If you decide not to define a navigation here then Zensical will simply
# derive the navigation structure from the directory structure of your
# "docs_dir". The definition below demonstrates how a navigation structure
# can be defined using TOML syntax.
#
# Read more: https://zensical.org/docs/setup/navigation/
# nav = [
# { "Get started" = "index.md" },
# { "Markdown in 5min" = "markdown.md" },
# ]
# With the "extra_css" option you can add your own CSS styling to customize
# your Zensical project according to your needs. You can add any number of
# CSS files.
#
# The path provided should be relative to the "docs_dir".
#
# Read more: https://zensical.org/docs/customization/#additional-css
#
#extra_css = ["stylesheets/extra.css"]
# With the `extra_javascript` option you can add your own JavaScript to your
# project to customize the behavior according to your needs.
#
# The path provided should be relative to the "docs_dir".
#
# Read more: https://zensical.org/docs/customization/#additional-javascript
#extra_javascript = ["javascripts/extra.js"]
# ----------------------------------------------------------------------------
# Section for configuring theme options
# ----------------------------------------------------------------------------
[project.theme]
# change this to "classic" to use the traditional Material for MkDocs look.
#variant = "classic"
# Zensical allows you to override specific blocks, partials, or whole
# templates as well as to define your own templates. To do this, uncomment
# the custom_dir setting below and set it to a directory in which you
# keep your template overrides.
#
# Read more:
# - https://zensical.org/docs/customization/#extending-the-theme
#
custom_dir = "docs/overrides"
# With the "favicon" option you can set your own image to use as the icon
# browsers will use in the browser title bar or tab bar. The path provided
# must be relative to the "docs_dir".
#
# Read more:
# - https://zensical.org/docs/setup/logo-and-icons/#favicon
# - https://developer.mozilla.org/en-US/docs/Glossary/Favicon
#
#favicon = "images/favicon.png"
# Zensical supports more than 60 different languages. This means that the
# labels and tooltips that Zensical's templates produce are translated.
# The "language" option allows you to set the language used. This language
# is also indicated in the HTML head element to help with accessibility
# and guide search engines and translation tools.
#
# The default language is "en" (English). It is possible to create
# sites with multiple languages and configure a language selector. See
# the documentation for details.
#
# Read more:
# - https://zensical.org/docs/setup/language/
#
language = "en"
# Zensical provides a number of feature toggles that change the behavior
# of the documentation site.
features = [
# Zensical includes an announcement bar. This feature allows users to
# dismiss it when they have read the announcement.
# https://zensical.org/docs/setup/header/#announcement-bar
"announce.dismiss",
# If you have a repository configured and turn on this feature, Zensical
# will generate an edit button for the page. This works for common
# repository hosting services.
# https://zensical.org/docs/setup/repository/#content-actions
#"content.action.edit",
# If you have a repository configured and turn on this feature, Zensical
# will generate a button that allows the user to view the Markdown
# code for the current page.
# https://zensical.org/docs/setup/repository/#content-actions
"content.action.view",
# Code annotations allow you to add an icon with a tooltip to your
# code blocks to provide explanations at crucial points.
# https://zensical.org/docs/authoring/code-blocks/#code-annotations
"content.code.annotate",
# This feature turns on a button in code blocks that allow users to
# copy the content to their clipboard without first selecting it.
# https://zensical.org/docs/authoring/code-blocks/#code-copy-button
"content.code.copy",
# Code blocks can include a button to allow for the selection of line
# ranges by the user.
# https://zensical.org/docs/authoring/code-blocks/#code-selection-button
"content.code.select",
# Zensical can render footnotes as inline tooltips, so the user can read
# the footnote without leaving the context of the document.
# https://zensical.org/docs/authoring/footnotes/#footnote-tooltips
"content.footnote.tooltips",
# If you have many content tabs that have the same titles (e.g., "Python",
# "JavaScript", "Cobol"), this feature causes all of them to switch to
# at the same time when the user chooses their language in one.
# https://zensical.org/docs/authoring/content-tabs/#linked-content-tabs
"content.tabs.link",
# With this feature enabled users can add tooltips to links that will be
# displayed when the mouse pointer hovers the link.
# https://zensical.org/docs/authoring/tooltips/#improved-tooltips
"content.tooltips",
# With this feature enabled, Zensical will automatically hide parts
# of the header when the user scrolls past a certain point.
# https://zensical.org/docs/setup/header/#automatic-hiding
# "header.autohide",
# Turn on this feature to expand all collapsible sections in the
# navigation sidebar by default.
# https://zensical.org/docs/setup/navigation/#navigation-expansion
# "navigation.expand",
# This feature turns on navigation elements in the footer that allow the
# user to navigate to a next or previous page.
# https://zensical.org/docs/setup/footer/#navigation
"navigation.footer",
# When section index pages are enabled, documents can be directly attached
# to sections, which is particularly useful for providing overview pages.
# https://zensical.org/docs/setup/navigation/#section-index-pages
"navigation.indexes",
# When instant navigation is enabled, clicks on all internal links will be
# intercepted and dispatched via XHR without fully reloading the page.
# https://zensical.org/docs/setup/navigation/#instant-navigation
"navigation.instant",
# With instant prefetching, your site will start to fetch a page once the
# user hovers over a link. This will reduce the perceived loading time
# for the user.
# https://zensical.org/docs/setup/navigation/#instant-prefetching
"navigation.instant.prefetch",
# In order to provide a better user experience on slow connections when
# using instant navigation, a progress indicator can be enabled.
# https://zensical.org/docs/setup/navigation/#progress-indicator
#"navigation.instant.progress",
# When navigation paths are activated, a breadcrumb navigation is rendered
# above the title of each page
# https://zensical.org/docs/setup/navigation/#navigation-path
"navigation.path",
# When pruning is enabled, only the visible navigation items are included
# in the rendered HTML, reducing the size of the built site by 33% or more.
# https://zensical.org/docs/setup/navigation/#navigation-pruning
#"navigation.prune",
# When sections are enabled, top-level sections are rendered as groups in
# the sidebar for viewports above 1220px, but remain as-is on mobile.
# https://zensical.org/docs/setup/navigation/#navigation-sections
"navigation.sections",
# When tabs are enabled, top-level sections are rendered in a menu layer
# below the header for viewports above 1220px, but remain as-is on mobile.
# https://zensical.org/docs/setup/navigation/#navigation-tabs
"navigation.tabs",
# When sticky tabs are enabled, navigation tabs will lock below the header
# and always remain visible when scrolling down.
# https://zensical.org/docs/setup/navigation/#sticky-navigation-tabs
#"navigation.tabs.sticky",
# A back-to-top button can be shown when the user, after scrolling down,
# starts to scroll up again.
# https://zensical.org/docs/setup/navigation/#back-to-top-button
"navigation.top",
# When anchor tracking is enabled, the URL in the address bar is
# automatically updated with the active anchor as highlighted in the table
# of contents.
# https://zensical.org/docs/setup/navigation/#anchor-tracking
"navigation.tracking",
# When search highlighting is enabled and a user clicks on a search result,
# Zensical will highlight all occurrences after following the link.
# https://zensical.org/docs/setup/search/#search-highlighting
"search.highlight",
# When anchor following for the table of contents is enabled, the sidebar
# is automatically scrolled so that the active anchor is always visible.
# https://zensical.org/docs/setup/navigation/#anchor-following
# "toc.follow",
# When navigation integration for the table of contents is enabled, it is
# always rendered as part of the navigation sidebar on the left.
# https://zensical.org/docs/setup/navigation/#navigation-integration
#"toc.integrate",
]
# ----------------------------------------------------------------------------
# In the "palette" subsection you can configure options for the color scheme.
# You can configure different color # schemes, e.g., to turn on dark mode,
# that the user can switch between. Each color scheme can be further
# customized.
#
# Read more:
# - https://zensical.org/docs/setup/colors/
# ----------------------------------------------------------------------------
[[project.theme.palette]]
scheme = "default"
toggle.icon = "lucide/sun"
@@ -270,43 +40,13 @@ scheme = "slate"
toggle.icon = "lucide/moon"
toggle.name = "Switch to light mode"
# ----------------------------------------------------------------------------
# In the "font" subsection you can configure the fonts used. By default, fonts
# are loaded from Google Fonts, giving you a wide range of choices from a set
# of suitably licensed fonts. There are options for a normal text font and for
# a monospaced font used in code blocks.
# ----------------------------------------------------------------------------
[project.theme.font]
text = "Inter"
code = "Jetbrains Mono"
# ----------------------------------------------------------------------------
# You can configure your own logo to be shown in the header using the "logo"
# option in the "icons" subsection. The logo can be a path to a file in your
# "docs_dir" or it can be a path to an icon.
#
# Likewise, you can customize the logo used for the repository section of the
# header. Zensical derives the default logo for this from the repository URL.
# See below...
#
# There are other icons you can customize. See the documentation for details.
#
# Read more:
# - https://zensical.org/docs/setup/logo-and-icons
# - https://zensical.org/docs/authoring/icons-emojis/#search
# ----------------------------------------------------------------------------
[project.theme.icon]
#logo = "lucide/smile"
repo = "fontawesome/brands/github"
# ----------------------------------------------------------------------------
# The "extra" section contains miscellaneous settings.
# ----------------------------------------------------------------------------
#[[project.extra.social]]
#icon = "fontawesome/brands/github"
#link = "https://github.com/user/repo"
[project.plugins.mkdocstrings.handlers.python]
inventories = ["https://docs.python.org/3/objects.inv"]
paths = ["src"]
@@ -316,3 +56,42 @@ docstring_style = "google"
inherited_members = true
show_source = false
show_root_heading = true
[project.markdown_extensions]
abbr = {}
admonition = {}
attr_list = {}
def_list = {}
footnotes = {}
md_in_html = {}
"pymdownx.arithmatex" = {generic = true}
"pymdownx.betterem" = {}
"pymdownx.caret" = {}
"pymdownx.details" = {}
"pymdownx.emoji" = {}
"pymdownx.inlinehilite" = {}
"pymdownx.keys" = {}
"pymdownx.magiclink" = {}
"pymdownx.mark" = {}
"pymdownx.smartsymbols" = {}
"pymdownx.tasklist" = {custom_checkbox = true}
"pymdownx.tilde" = {}
[project.markdown_extensions."pymdownx.highlight"]
anchor_linenums = true
line_spans = "__span"
pygments_lang_class = true
[project.markdown_extensions."pymdownx.superfences"]
custom_fences = [{name = "mermaid", class = "mermaid"}]
[project.markdown_extensions."pymdownx.tabbed"]
alternate_style = true
combine_header_slug = true
[project.markdown_extensions."toc"]
permalink = true
[project.markdown_extensions."pymdownx.snippets"]
base_path = ["."]
check_paths = true