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 ### 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 - **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 - **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 - **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 ### 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 - **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 - **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 - **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 ### Cursor pagination
```python ```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). 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 | | Input | Free-text string | Exact column values |
| Relationship support | Yes | Yes | | Relationship support | Yes | Yes |
@@ -242,7 +239,7 @@ async def get_users(
) )
``` ```
### Filter attributes ### Faceted search
!!! info "Added in `v1.2`" !!! info "Added in `v1.2`"
@@ -384,7 +381,7 @@ await UserCrud.upsert(
) )
``` ```
## `schema` — typed response serialization ## Response serialization
!!! info "Added in `v1.1`" !!! 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. 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) [: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] [project]
name = "fastapi-toolsets" name = "fastapi-toolsets"
version = "1.2.0" version = "1.2.1"
description = "Reusable tools for FastAPI: async CRUD, fixtures, CLI, and standardized responses for SQLAlchemy + PostgreSQL" description = "Production-ready utilities for FastAPI applications"
readme = "README.md" readme = "README.md"
license = "MIT" license = "MIT"
license-files = ["LICENSE"] license-files = ["LICENSE"]
@@ -11,7 +11,7 @@ authors = [
] ]
keywords = ["fastapi", "sqlalchemy", "postgresql"] keywords = ["fastapi", "sqlalchemy", "postgresql"]
classifiers = [ classifiers = [
"Development Status :: 4 - Beta", "Development Status :: 5 - Production/Stable",
"Framework :: AsyncIO", "Framework :: AsyncIO",
"Framework :: FastAPI", "Framework :: FastAPI",
"Framework :: Pydantic", "Framework :: Pydantic",

View File

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

View File

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

View File

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

View File

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

View File

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

2
uv.lock generated
View File

@@ -251,7 +251,7 @@ wheels = [
[[package]] [[package]]
name = "fastapi-toolsets" name = "fastapi-toolsets"
version = "1.2.0" version = "1.2.1"
source = { editable = "." } source = { editable = "." }
dependencies = [ dependencies = [
{ name = "asyncpg" }, { 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] [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" 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." 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" 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" site_url = "https://fastapi-toolsets.d3vyce.fr"
copyright = "Copyright &copy; 2026 d3vyce"
# 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
"""
repo_url = "https://github.com/d3vyce/fastapi-toolsets" 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] [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" 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" language = "en"
# Zensical provides a number of feature toggles that change the behavior
# of the documentation site.
features = [ 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", "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", "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", "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", "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", "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", "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", "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", "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", "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", "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", "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", "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", "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", "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", "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", "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", "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", "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]] [[project.theme.palette]]
scheme = "default" scheme = "default"
toggle.icon = "lucide/sun" toggle.icon = "lucide/sun"
@@ -270,43 +40,13 @@ scheme = "slate"
toggle.icon = "lucide/moon" toggle.icon = "lucide/moon"
toggle.name = "Switch to light mode" 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] [project.theme.font]
text = "Inter" text = "Inter"
code = "Jetbrains Mono" 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] [project.theme.icon]
#logo = "lucide/smile"
repo = "fontawesome/brands/github" 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] [project.plugins.mkdocstrings.handlers.python]
inventories = ["https://docs.python.org/3/objects.inv"] inventories = ["https://docs.python.org/3/objects.inv"]
paths = ["src"] paths = ["src"]
@@ -316,3 +56,42 @@ docstring_style = "google"
inherited_members = true inherited_members = true
show_source = false show_source = false
show_root_heading = true 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