20 Commits
v3.0.0 ... main

Author SHA1 Message Date
94e7d79d06 Version 3.1.0 2026-04-12 12:48:32 -04:00
d3vyce
9268b576b4 feat: add M2M helpers (#247) 2026-04-12 18:46:57 +02:00
d3vyce
863e6ce6e9 feat: add get_field_by_attr fixtures helper (#245) 2026-04-12 17:12:08 +02:00
d3vyce
c7397faea4 feat: auto eager-load relationships in register_fixtures (#243) 2026-04-12 17:04:44 +02:00
dependabot[bot]
aca3f62a6b ⬆ Bump ruff from 0.15.8 to 0.15.9 (#241)
Bumps [ruff](https://github.com/astral-sh/ruff) from 0.15.8 to 0.15.9.
- [Release notes](https://github.com/astral-sh/ruff/releases)
- [Changelog](https://github.com/astral-sh/ruff/blob/main/CHANGELOG.md)
- [Commits](https://github.com/astral-sh/ruff/compare/0.15.8...0.15.9)

---
updated-dependencies:
- dependency-name: ruff
  dependency-version: 0.15.9
  dependency-type: direct:development
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-04-09 10:01:30 +02:00
dependabot[bot]
8030e1988c ⬆ Bump zensical from 0.0.31 to 0.0.32 (#240)
Bumps [zensical](https://github.com/zensical/zensical) from 0.0.31 to 0.0.32.
- [Release notes](https://github.com/zensical/zensical/releases)
- [Commits](https://github.com/zensical/zensical/compare/v0.0.31...v0.0.32)

---
updated-dependencies:
- dependency-name: zensical
  dependency-version: 0.0.32
  dependency-type: direct:development
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-04-09 10:01:14 +02:00
dependabot[bot]
e2c2c1c835 ⬆ Bump ty from 0.0.27 to 0.0.29 (#239)
Bumps [ty](https://github.com/astral-sh/ty) from 0.0.27 to 0.0.29.
- [Release notes](https://github.com/astral-sh/ty/releases)
- [Changelog](https://github.com/astral-sh/ty/blob/main/CHANGELOG.md)
- [Commits](https://github.com/astral-sh/ty/compare/0.0.27...0.0.29)

---
updated-dependencies:
- dependency-name: ty
  dependency-version: 0.0.29
  dependency-type: direct:development
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-04-09 10:01:02 +02:00
dependabot[bot]
025b954d01 ⬆ Bump pytest from 9.0.2 to 9.0.3 (#238)
Bumps [pytest](https://github.com/pytest-dev/pytest) from 9.0.2 to 9.0.3.
- [Release notes](https://github.com/pytest-dev/pytest/releases)
- [Changelog](https://github.com/pytest-dev/pytest/blob/main/CHANGELOG.rst)
- [Commits](https://github.com/pytest-dev/pytest/compare/9.0.2...9.0.3)

---
updated-dependencies:
- dependency-name: pytest
  dependency-version: 9.0.3
  dependency-type: direct:development
  update-type: version-update:semver-patch
...

Signed-off-by: dependabot[bot] <support@github.com>
Co-authored-by: dependabot[bot] <49699333+dependabot[bot]@users.noreply.github.com>
2026-04-09 10:00:49 +02:00
0ed93d62c8 Version 3.0.3 2026-04-07 05:43:10 -04:00
d3vyce
2a49814818 feat: normalize enum facet values to member names and accept name strings in filter_by (#235) 2026-04-07 11:42:24 +02:00
d3vyce
f8e090c7c3 ci: restrict ci.yml workflow permissions to contents: read (#233) 2026-04-07 10:44:59 +02:00
d3vyce
54decaf3e1 fix: coerce plain int values to enum member when filtering on IntEnum column (#231) 2026-04-07 10:43:01 +02:00
6b127d9645 Version 3.0.2 2026-04-04 19:13:09 -04:00
d3vyce
8bed96f4bf fix: apply default_load_options after create() and update() to prevent MissingGreenlet on relationship access (#229) 2026-04-04 20:47:05 +02:00
d3vyce
74d15e13bc feat: support relation tuples in order_fields for cross-table sorting (#227)
* feat: support relation tuples in order_fields for cross-table sorting

* docs: update crud module
2026-04-04 17:00:14 +02:00
d3vyce
e38d8d2d4f feat: expose sort_columns in paginated response (#225) 2026-04-04 14:46:36 +02:00
9b74f162ab Version 3.0.1 2026-04-03 05:44:10 -04:00
d3vyce
ab125c6ea1 docs: rework versioning to keep latest feature version per major (#223) 2026-04-03 11:43:36 +02:00
d3vyce
e388e26858 fix: widen JoinType to accept aliased and polymorphic targets (#221) 2026-04-02 22:58:52 +02:00
d3vyce
04da241294 fix: coerce string values to bool for Boolean facet field filtering (#219) 2026-04-02 22:58:07 +02:00
24 changed files with 2098 additions and 175 deletions

View File

@@ -6,6 +6,9 @@ on:
pull_request:
branches: [main]
permissions:
contents: read
concurrency:
group: ${{ github.workflow }}-${{ github.ref }}
cancel-in-progress: true

View File

@@ -34,26 +34,17 @@ jobs:
MAJOR=$(echo "$VERSION" | cut -d. -f1)
DEPLOY_VERSION="v$(echo "$VERSION" | cut -d. -f1-2)"
# On new major: consolidate previous major's feature versions into vX
# On new major: keep only the latest feature version of the previous major
PREV_MAJOR=$((MAJOR - 1))
OLD_FEATURE_VERSIONS=$(uv run mike list 2>/dev/null | grep -oE "^v${PREV_MAJOR}\.[0-9]+" || true)
if [ -n "$OLD_FEATURE_VERSIONS" ]; then
LATEST_PREV_TAG=$(git tag -l "v${PREV_MAJOR}.*" | sort -V | tail -1)
if [ -n "$LATEST_PREV_TAG" ]; then
git checkout "$LATEST_PREV_TAG" -- docs/ docs_src/ src/ zensical.toml
if ! grep -q '\[project\.extra\.version\]' zensical.toml; then
printf '\n[project.extra.version]\nprovider = "mike"\ndefault = "stable"\nalias = true\n' >> zensical.toml
fi
uv run mike deploy "v${PREV_MAJOR}"
git checkout HEAD -- docs/ docs_src/ src/ zensical.toml
fi
# Delete old feature versions
LATEST_PREV=$(echo "$OLD_FEATURE_VERSIONS" | sort -t. -k2 -n | tail -1)
echo "$OLD_FEATURE_VERSIONS" | while read -r OLD_V; do
echo "Deleting $OLD_V"
uv run mike delete "$OLD_V"
if [ "$OLD_V" != "$LATEST_PREV" ]; then
echo "Deleting $OLD_V"
uv run mike delete "$OLD_V"
fi
done
fi

View File

@@ -324,6 +324,12 @@ result = await UserCrud.offset_paginate(
)
```
Or via the dependency to narrow which fields are exposed as query parameters:
```python
params = UserCrud.offset_paginate_params(search_fields=[Post.title])
```
This allows searching with both [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) and [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate):
```python
@@ -344,13 +350,37 @@ async def get_users(
return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)
```
The dependency adds two query parameters to the endpoint:
| Parameter | Type |
| --------------- | ------------- |
| `search` | `str \| null` |
| `search_column` | `str \| null` |
```
GET /posts?search=hello → search all configured columns
GET /posts?search=hello&search_column=title → search only Post.title
```
The available search column keys are returned in the `search_columns` field of [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse). Use them to populate a column picker in the UI, or to validate `search_column` values on the client side:
```json
{
"status": "SUCCESS",
"data": ["..."],
"pagination": { "..." },
"search_columns": ["content", "author__username", "title"]
}
```
!!! info "Key format uses `__` as a separator for relationship chains."
A direct column `Post.title` produces `"title"`. A relationship tuple `(Post.author, User.username)` produces `"author__username"`. An unknown `search_column` value raises [`InvalidSearchColumnError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidSearchColumnError) (HTTP 422).
### Faceted search
!!! info "Added in `v1.2`"
Declare `facet_fields` on the CRUD class to return distinct column values alongside paginated results. This is useful for populating filter dropdowns or building faceted search UIs.
Facet fields use the same syntax as `searchable_fields` — direct columns or relationship tuples:
Declare `facet_fields` on the CRUD class to return distinct column values alongside paginated results. This is useful for populating filter dropdowns or building faceted search UIs. Relationship traversal is supported via tuples, using the same syntax as `searchable_fields`:
```python
UserCrud = CrudFactory(
@@ -372,7 +402,47 @@ result = await UserCrud.offset_paginate(
)
```
The distinct values are returned in the `filter_attributes` field of [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse):
Or via the dependency to narrow which fields are exposed as query parameters:
```python
params = UserCrud.offset_paginate_params(facet_fields=[User.country])
```
Facet filtering is built into the consolidated params dependencies. When `filter=True` (the default), each facet field is exposed as a query parameter and values are collected into `filter_by` automatically:
```python
from typing import Annotated
from fastapi import Depends
@router.get("", response_model_exclude_none=True)
async def list_users(
session: SessionDep,
params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
) -> OffsetPaginatedResponse[UserRead]:
return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)
```
```python
@router.get("", response_model_exclude_none=True)
async def list_users(
session: SessionDep,
params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())],
) -> CursorPaginatedResponse[UserRead]:
return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)
```
Both single-value and multi-value query parameters work:
```
GET /users?status=active → filter_by={"status": ["active"]}
GET /users?status=active&country=FR → filter_by={"status": ["active"], "country": ["FR"]}
GET /users?role__name=admin&role__name=editor → filter_by={"role__name": ["admin", "editor"]} (IN clause)
```
`filter_by` and `filters` can be combined — both are applied with AND logic.
The distinct values for each facet field are returned in the `filter_attributes` field of [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse). Use them to populate filter dropdowns in the UI, or to validate `filter_by` keys on the client side:
```json
{
@@ -387,50 +457,14 @@ The distinct values are returned in the `filter_attributes` field of [`Paginated
}
```
Use `filter_by` to pass the client's chosen filter values directly — no need to build SQLAlchemy conditions by hand. Any unknown key raises [`InvalidFacetFilterError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidFacetFilterError).
!!! info "The keys in `filter_by` are the same keys the client received in `filter_attributes`."
Keys use `__` as a separator for the full relationship chain. A direct column `User.status` produces `"status"`. A relationship tuple `(User.role, Role.name)` produces `"role__name"`. A deeper chain `(User.role, Role.permission, Permission.name)` produces `"role__permission__name"`.
`filter_by` and `filters` can be combined — both are applied with AND logic.
Facet filtering is built into the consolidated params dependencies. When `filter=True` (the default), facet fields are exposed as query parameters and collected into `filter_by` automatically:
```python
from typing import Annotated
from fastapi import Depends
UserCrud = CrudFactory(
model=User,
facet_fields=[User.status, User.country, (User.role, Role.name)],
)
@router.get("", response_model_exclude_none=True)
async def list_users(
session: SessionDep,
params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
) -> OffsetPaginatedResponse[UserRead]:
return await UserCrud.offset_paginate(
session=session,
**params,
schema=UserRead,
)
```
Both single-value and multi-value query parameters work:
```
GET /users?status=active → filter_by={"status": ["active"]}
GET /users?status=active&country=FR → filter_by={"status": ["active"], "country": ["FR"]}
GET /users?role__name=admin&role__name=editor → filter_by={"role__name": ["admin", "editor"]} (IN clause)
```
!!! info "Key format uses `__` as a separator for relationship chains."
A direct column `User.status` produces `"status"`. A relationship tuple `(User.role, Role.name)` produces `"role__name"`. A deeper chain `(User.role, Role.permission, Permission.name)` produces `"role__permission__name"`. An unknown `filter_by` key raises [`InvalidFacetFilterError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidFacetFilterError) (HTTP 422).
## Sorting
!!! info "Added in `v1.3`"
Declare `order_fields` on the CRUD class to expose client-driven column ordering via `order_by` and `order` query parameters.
Declare `order_fields` on the CRUD class. Relationship traversal is supported via tuples, using the same syntax as `searchable_fields` and `facet_fields`:
```python
UserCrud = CrudFactory(
@@ -438,11 +472,27 @@ UserCrud = CrudFactory(
order_fields=[
User.name,
User.created_at,
(User.role, Role.name), # sort by a related model column
],
)
```
Ordering is built into the consolidated params dependencies. When `order=True` (the default), `order_by` and `order` query parameters are exposed and resolved into an `OrderByClause` automatically:
You can override `order_fields` per call:
```python
result = await UserCrud.offset_paginate(
session=session,
order_fields=[User.name],
)
```
Or via the dependency to narrow which fields are exposed as query parameters:
```python
params = UserCrud.offset_paginate_params(order_fields=[User.name])
```
Sorting is built into the consolidated params dependencies. When `order=True` (the default), `order_by` and `order` query parameters are exposed and resolved into an `OrderByClause` automatically:
```python
from typing import Annotated
@@ -452,33 +502,50 @@ from fastapi import Depends
@router.get("")
async def list_users(
session: SessionDep,
params: Annotated[dict, Depends(UserCrud.offset_paginate_params(
default_order_field=User.created_at,
))],
params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
) -> OffsetPaginatedResponse[UserRead]:
return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)
```
```python
@router.get("")
async def list_users(
session: SessionDep,
params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())],
) -> CursorPaginatedResponse[UserRead]:
return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)
```
The dependency adds two query parameters to the endpoint:
| Parameter | Type |
| ---------- | --------------- |
| `order_by` | `str | null` |
| `order_by` | `str \| null` |
| `order` | `asc` or `desc` |
```
GET /users?order_by=name&order=asc → ORDER BY users.name ASC
GET /users?order_by=name&order=desc → ORDER BY users.name DESC
GET /users?order_by=name&order=asc → ORDER BY users.name ASC
GET /users?order_by=role__name&order=desc → LEFT JOIN roles ON ... ORDER BY roles.name DESC
```
An unknown `order_by` value raises [`InvalidOrderFieldError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidOrderFieldError) (HTTP 422).
!!! info "Relationship tuples are joined automatically."
When a relation field is selected, the related table is LEFT OUTER JOINed automatically. An unknown `order_by` value raises [`InvalidOrderFieldError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidOrderFieldError) (HTTP 422).
You can also pass `order_fields` directly to override the class-level defaults:
```python
params = UserCrud.offset_paginate_params(order_fields=[User.name])
The available sort keys are returned in the `order_columns` field of [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse). Use them to populate a sort picker in the UI, or to validate `order_by` values on the client side:
```json
{
"status": "SUCCESS",
"data": ["..."],
"pagination": { "..." },
"order_columns": ["created_at", "name", "role__name"]
}
```
!!! info "Key format uses `__` as a separator for relationship chains."
A direct column `User.name` produces `"name"`. A relationship tuple `(User.role, Role.name)` produces `"role__name"`.
## Relationship loading
!!! info "Added in `v1.1`"

View File

@@ -118,6 +118,57 @@ async def clean(db_session):
await cleanup_tables(session=db_session, base=Base)
```
## Many-to-Many helpers
SQLAlchemy's ORM collection API triggers lazy-loads when you append to a relationship inside a savepoint (e.g. inside `lock_tables` or a nested `get_transaction`). The three `m2m_*` helpers bypass the ORM collection entirely and issue direct SQL against the association table.
### `m2m_add` — insert associations
[`m2m_add`](../reference/db.md#fastapi_toolsets.db.m2m_add) inserts one or more rows into a secondary table without touching the ORM collection:
```python
from fastapi_toolsets.db import lock_tables, m2m_add
async with lock_tables(session, [Tag]):
tag = await TagCrud.create(session, TagCreate(name="python"))
await m2m_add(session, post, Post.tags, tag)
```
Pass `ignore_conflicts=True` to silently skip associations that already exist:
```python
await m2m_add(session, post, Post.tags, tag, ignore_conflicts=True)
```
### `m2m_remove` — delete associations
[`m2m_remove`](../reference/db.md#fastapi_toolsets.db.m2m_remove) deletes specific association rows. Removing a non-existent association is a no-op:
```python
from fastapi_toolsets.db import get_transaction, m2m_remove
async with get_transaction(session):
await m2m_remove(session, post, Post.tags, tag1, tag2)
```
### `m2m_set` — replace the full set
[`m2m_set`](../reference/db.md#fastapi_toolsets.db.m2m_set) atomically replaces all associations: it deletes every existing row for the owner instance then inserts the new set. Passing no related instances clears the association entirely:
```python
from fastapi_toolsets.db import get_transaction, m2m_set
# Replace all tags
async with get_transaction(session):
await m2m_set(session, post, Post.tags, tag_a, tag_b)
# Clear all tags
async with get_transaction(session):
await m2m_set(session, post, Post.tags)
```
All three helpers raise `TypeError` if the relationship attribute is not a Many-to-Many (i.e. has no secondary table).
---
[:material-api: API Reference](../reference/db.md)

View File

@@ -13,6 +13,9 @@ from fastapi_toolsets.db import (
create_db_context,
get_transaction,
lock_tables,
m2m_add,
m2m_remove,
m2m_set,
wait_for_row_change,
)
```
@@ -32,3 +35,9 @@ from fastapi_toolsets.db import (
## ::: fastapi_toolsets.db.create_database
## ::: fastapi_toolsets.db.cleanup_tables
## ::: fastapi_toolsets.db.m2m_add
## ::: fastapi_toolsets.db.m2m_remove
## ::: fastapi_toolsets.db.m2m_set

View File

@@ -1,6 +1,6 @@
[project]
name = "fastapi-toolsets"
version = "3.0.0"
version = "3.1.0"
description = "Production-ready utilities for FastAPI applications"
readme = "README.md"
license = "MIT"

View File

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

View File

@@ -12,6 +12,7 @@ from ..types import (
JoinType,
M2MFieldType,
OrderByClause,
OrderFieldType,
SearchFieldType,
)
from .factory import AsyncCrud, CrudFactory
@@ -28,6 +29,7 @@ __all__ = [
"M2MFieldType",
"NoSearchableFieldsError",
"OrderByClause",
"OrderFieldType",
"PaginationType",
"SearchConfig",
"SearchFieldType",

View File

@@ -38,6 +38,7 @@ from ..types import (
M2MFieldType,
ModelType,
OrderByClause,
OrderFieldType,
SchemaType,
SearchFieldType,
)
@@ -134,7 +135,7 @@ class AsyncCrud(Generic[ModelType]):
model: ClassVar[type[DeclarativeBase]]
searchable_fields: ClassVar[Sequence[SearchFieldType] | None] = None
facet_fields: ClassVar[Sequence[FacetFieldType] | None] = None
order_fields: ClassVar[Sequence[QueryableAttribute[Any]] | None] = None
order_fields: ClassVar[Sequence[OrderFieldType] | None] = None
m2m_fields: ClassVar[M2MFieldType | None] = None
default_load_options: ClassVar[Sequence[ExecutableOption] | None] = None
cursor_column: ClassVar[Any | None] = None
@@ -169,6 +170,18 @@ class AsyncCrud(Generic[ModelType]):
return load_options
return cls.default_load_options
@classmethod
async def _reload_with_options(
cls: type[Self], session: AsyncSession, instance: ModelType
) -> ModelType:
"""Re-query instance by PK with default_load_options applied."""
mapper = cls.model.__mapper__
pk_filters = [
getattr(cls.model, col.key) == getattr(instance, col.key)
for col in mapper.primary_key
]
return await cls.get(session, filters=pk_filters)
@classmethod
async def _resolve_m2m(
cls: type[Self],
@@ -278,6 +291,17 @@ class AsyncCrud(Generic[ModelType]):
return None
return search_field_keys(fields)
@classmethod
def _resolve_order_columns(
cls: type[Self],
order_fields: Sequence[OrderFieldType] | None,
) -> list[str] | None:
"""Return sort column keys, or None if no order fields configured."""
fields = order_fields if order_fields is not None else cls.order_fields
if not fields:
return None
return sorted(facet_keys(fields))
@classmethod
def _build_paginate_params(
cls: type[Self],
@@ -290,7 +314,7 @@ class AsyncCrud(Generic[ModelType]):
order: bool,
search_fields: Sequence[SearchFieldType] | None,
facet_fields: Sequence[FacetFieldType] | None,
order_fields: Sequence[QueryableAttribute[Any]] | None,
order_fields: Sequence[OrderFieldType] | None,
default_order_field: QueryableAttribute[Any] | None,
default_order: Literal["asc", "desc"],
) -> Callable[..., Awaitable[dict[str, Any]]]:
@@ -349,14 +373,15 @@ class AsyncCrud(Generic[ModelType]):
)
reserved_names.update(filter_keys)
order_field_map: dict[str, QueryableAttribute[Any]] | None = None
order_field_map: dict[str, OrderFieldType] | None = None
order_valid_keys: list[str] | None = None
if order:
resolved_order = (
order_fields if order_fields is not None else cls.order_fields
)
if resolved_order:
order_field_map = {f.key: f for f in resolved_order}
keys = facet_keys(resolved_order)
order_field_map = dict(zip(keys, resolved_order))
order_valid_keys = sorted(order_field_map.keys())
all_params.extend(
[
@@ -408,9 +433,16 @@ class AsyncCrud(Generic[ModelType]):
else:
field = order_field_map[order_by_val]
if field is not None:
result["order_by"] = (
field.asc() if order_dir == "asc" else field.desc()
)
if isinstance(field, tuple):
col = field[-1]
result["order_by"] = (
col.asc() if order_dir == "asc" else col.desc()
)
result["order_joins"] = list(field[:-1])
else:
result["order_by"] = (
field.asc() if order_dir == "asc" else field.desc()
)
else:
result["order_by"] = None
@@ -434,7 +466,7 @@ class AsyncCrud(Generic[ModelType]):
order: bool = True,
search_fields: Sequence[SearchFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
order_fields: Sequence[OrderFieldType] | None = None,
default_order_field: QueryableAttribute[Any] | None = None,
default_order: Literal["asc", "desc"] = "asc",
) -> Callable[..., Awaitable[dict[str, Any]]]:
@@ -496,7 +528,7 @@ class AsyncCrud(Generic[ModelType]):
order: bool = True,
search_fields: Sequence[SearchFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
order_fields: Sequence[OrderFieldType] | None = None,
default_order_field: QueryableAttribute[Any] | None = None,
default_order: Literal["asc", "desc"] = "asc",
) -> Callable[..., Awaitable[dict[str, Any]]]:
@@ -561,7 +593,7 @@ class AsyncCrud(Generic[ModelType]):
order: bool = True,
search_fields: Sequence[SearchFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
order_fields: Sequence[OrderFieldType] | None = None,
default_order_field: QueryableAttribute[Any] | None = None,
default_order: Literal["asc", "desc"] = "asc",
) -> Callable[..., Awaitable[dict[str, Any]]]:
@@ -685,6 +717,8 @@ class AsyncCrud(Generic[ModelType]):
session.add(db_model)
await session.refresh(db_model)
if cls.default_load_options:
db_model = await cls._reload_with_options(session, db_model)
result = cast(ModelType, db_model)
if schema:
return Response(data=schema.model_validate(result))
@@ -1040,6 +1074,8 @@ class AsyncCrud(Generic[ModelType]):
for rel_attr, related_instances in m2m_resolved.items():
setattr(db_model, rel_attr, related_instances)
await session.refresh(db_model)
if cls.default_load_options:
db_model = await cls._reload_with_options(session, db_model)
if schema:
return Response(data=schema.model_validate(db_model))
return db_model
@@ -1202,12 +1238,14 @@ class AsyncCrud(Generic[ModelType]):
outer_join: bool = False,
load_options: Sequence[ExecutableOption] | None = None,
order_by: OrderByClause | None = None,
order_joins: list[Any] | None = None,
page: int = 1,
items_per_page: int = 20,
include_total: bool = True,
search: str | SearchConfig | None = None,
search_fields: Sequence[SearchFieldType] | None = None,
search_column: str | None = None,
order_fields: Sequence[OrderFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel],
@@ -1228,6 +1266,7 @@ class AsyncCrud(Generic[ModelType]):
search: Search query string or SearchConfig object
search_fields: Fields to search in (overrides class default)
search_column: Restrict search to a single column key.
order_fields: Fields allowed for sorting (overrides class default).
facet_fields: Columns to compute distinct values for (overrides class default)
filter_by: Dict of {column_key: value} to filter by declared facet fields.
Keys must match the column.key of a facet field. Scalar → equality,
@@ -1264,6 +1303,10 @@ class AsyncCrud(Generic[ModelType]):
# Apply search joins (always outer joins for search)
q = _apply_search_joins(q, search_joins)
# Apply order joins (relation joins required for order_by field)
if order_joins:
q = _apply_search_joins(q, order_joins)
if filters:
q = q.where(and_(*filters))
if resolved := cls._resolve_load_options(load_options):
@@ -1308,6 +1351,7 @@ class AsyncCrud(Generic[ModelType]):
session, facet_fields, filters, search_joins
)
search_columns = cls._resolve_search_columns(search_fields)
order_columns = cls._resolve_order_columns(order_fields)
return OffsetPaginatedResponse(
data=items,
@@ -1319,6 +1363,7 @@ class AsyncCrud(Generic[ModelType]):
),
filter_attributes=filter_attributes,
search_columns=search_columns,
order_columns=order_columns,
)
@classmethod
@@ -1332,10 +1377,12 @@ class AsyncCrud(Generic[ModelType]):
outer_join: bool = False,
load_options: Sequence[ExecutableOption] | None = None,
order_by: OrderByClause | None = None,
order_joins: list[Any] | None = None,
items_per_page: int = 20,
search: str | SearchConfig | None = None,
search_fields: Sequence[SearchFieldType] | None = None,
search_column: str | None = None,
order_fields: Sequence[OrderFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel],
@@ -1357,6 +1404,7 @@ class AsyncCrud(Generic[ModelType]):
search: Search query string or SearchConfig object.
search_fields: Fields to search in (overrides class default).
search_column: Restrict search to a single column key.
order_fields: Fields allowed for sorting (overrides class default).
facet_fields: Columns to compute distinct values for (overrides class default).
filter_by: Dict of {column_key: value} to filter by declared facet fields.
Keys must match the column.key of a facet field. Scalar → equality,
@@ -1410,6 +1458,10 @@ class AsyncCrud(Generic[ModelType]):
# Apply search joins (always outer joins)
q = _apply_search_joins(q, search_joins)
# Apply order joins (relation joins required for order_by field)
if order_joins:
q = _apply_search_joins(q, order_joins)
if filters:
q = q.where(and_(*filters))
if resolved := cls._resolve_load_options(load_options):
@@ -1468,6 +1520,7 @@ class AsyncCrud(Generic[ModelType]):
session, facet_fields, filters, search_joins
)
search_columns = cls._resolve_search_columns(search_fields)
order_columns = cls._resolve_order_columns(order_fields)
return CursorPaginatedResponse(
data=items,
@@ -1479,6 +1532,7 @@ class AsyncCrud(Generic[ModelType]):
),
filter_attributes=filter_attributes,
search_columns=search_columns,
order_columns=order_columns,
)
@overload
@@ -1493,6 +1547,7 @@ class AsyncCrud(Generic[ModelType]):
outer_join: bool = ...,
load_options: Sequence[ExecutableOption] | None = ...,
order_by: OrderByClause | None = ...,
order_joins: list[Any] | None = ...,
page: int = ...,
cursor: str | None = ...,
items_per_page: int = ...,
@@ -1500,6 +1555,7 @@ class AsyncCrud(Generic[ModelType]):
search: str | SearchConfig | None = ...,
search_fields: Sequence[SearchFieldType] | None = ...,
search_column: str | None = ...,
order_fields: Sequence[OrderFieldType] | None = ...,
facet_fields: Sequence[FacetFieldType] | None = ...,
filter_by: dict[str, Any] | BaseModel | None = ...,
schema: type[BaseModel],
@@ -1517,6 +1573,7 @@ class AsyncCrud(Generic[ModelType]):
outer_join: bool = ...,
load_options: Sequence[ExecutableOption] | None = ...,
order_by: OrderByClause | None = ...,
order_joins: list[Any] | None = ...,
page: int = ...,
cursor: str | None = ...,
items_per_page: int = ...,
@@ -1524,6 +1581,7 @@ class AsyncCrud(Generic[ModelType]):
search: str | SearchConfig | None = ...,
search_fields: Sequence[SearchFieldType] | None = ...,
search_column: str | None = ...,
order_fields: Sequence[OrderFieldType] | None = ...,
facet_fields: Sequence[FacetFieldType] | None = ...,
filter_by: dict[str, Any] | BaseModel | None = ...,
schema: type[BaseModel],
@@ -1540,6 +1598,7 @@ class AsyncCrud(Generic[ModelType]):
outer_join: bool = False,
load_options: Sequence[ExecutableOption] | None = None,
order_by: OrderByClause | None = None,
order_joins: list[Any] | None = None,
page: int = 1,
cursor: str | None = None,
items_per_page: int = 20,
@@ -1547,6 +1606,7 @@ class AsyncCrud(Generic[ModelType]):
search: str | SearchConfig | None = None,
search_fields: Sequence[SearchFieldType] | None = None,
search_column: str | None = None,
order_fields: Sequence[OrderFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
filter_by: dict[str, Any] | BaseModel | None = None,
schema: type[BaseModel],
@@ -1575,6 +1635,7 @@ class AsyncCrud(Generic[ModelType]):
search: Search query string or :class:`.SearchConfig` object.
search_fields: Fields to search in (overrides class default).
search_column: Restrict search to a single column key.
order_fields: Fields allowed for sorting (overrides class default).
facet_fields: Columns to compute distinct values for (overrides
class default).
filter_by: Dict of ``{column_key: value}`` to filter by declared
@@ -1600,10 +1661,12 @@ class AsyncCrud(Generic[ModelType]):
outer_join=outer_join,
load_options=load_options,
order_by=order_by,
order_joins=order_joins,
items_per_page=items_per_page,
search=search,
search_fields=search_fields,
search_column=search_column,
order_fields=order_fields,
facet_fields=facet_fields,
filter_by=filter_by,
schema=schema,
@@ -1618,12 +1681,14 @@ class AsyncCrud(Generic[ModelType]):
outer_join=outer_join,
load_options=load_options,
order_by=order_by,
order_joins=order_joins,
page=page,
items_per_page=items_per_page,
include_total=include_total,
search=search,
search_fields=search_fields,
search_column=search_column,
order_fields=order_fields,
facet_fields=facet_fields,
filter_by=filter_by,
schema=schema,
@@ -1638,7 +1703,7 @@ def CrudFactory(
base_class: type[AsyncCrud[Any]] = AsyncCrud,
searchable_fields: Sequence[SearchFieldType] | None = None,
facet_fields: Sequence[FacetFieldType] | None = None,
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
order_fields: Sequence[OrderFieldType] | None = None,
m2m_fields: M2MFieldType | None = None,
default_load_options: Sequence[ExecutableOption] | None = None,
cursor_column: Any | None = None,

View File

@@ -265,7 +265,15 @@ async def build_facets(
else:
q = q.order_by(column)
result = await session.execute(q)
values = [row[0] for row in result.all() if row[0] is not None]
col_type = column.property.columns[0].type
enum_class = getattr(col_type, "enum_class", None)
values = [
row[0].name
if (enum_class is not None and isinstance(row[0], enum_class))
else row[0]
for row in result.all()
if row[0] is not None
]
return key, values
pairs = await asyncio.gather(
@@ -278,6 +286,18 @@ _EQUALITY_TYPES = (String, Integer, Numeric, Date, DateTime, Time, Enum, Uuid)
"""Column types that support equality / IN filtering in build_filter_by."""
def _coerce_bool(value: Any) -> bool:
"""Coerce a string value to a Python bool for Boolean column filtering."""
if isinstance(value, bool):
return value
if isinstance(value, str):
if value.lower() == "true":
return True
if value.lower() == "false":
return False
raise ValueError(f"Cannot coerce {value!r} to bool")
def build_filter_by(
filter_by: dict[str, Any],
facet_fields: Sequence[FacetFieldType],
@@ -324,16 +344,35 @@ def build_filter_by(
added_join_keys.add(rel_key)
col_type = column.property.columns[0].type
if isinstance(col_type, ARRAY):
if isinstance(col_type, Boolean):
coerce = _coerce_bool
if isinstance(value, list):
filters.append(column.in_([coerce(v) for v in value]))
else:
filters.append(column == coerce(value))
elif isinstance(col_type, ARRAY):
if isinstance(value, list):
filters.append(column.overlap(value))
else:
filters.append(column.any(value))
elif isinstance(col_type, Boolean):
if isinstance(value, list):
filters.append(column.in_(value))
else:
filters.append(column.is_(value))
elif isinstance(col_type, Enum):
enum_class = col_type.enum_class
if enum_class is not None:
def _coerce_enum(v: Any) -> Any:
if isinstance(v, enum_class):
return v
return enum_class[v] # lookup by name: "PENDING", "RED"
if isinstance(value, list):
filters.append(column.in_([_coerce_enum(v) for v in value]))
else:
filters.append(column == _coerce_enum(value))
else: # pragma: no cover
if isinstance(value, list):
filters.append(column.in_(value))
else:
filters.append(column == value)
elif isinstance(col_type, _EQUALITY_TYPES):
if isinstance(value, list):
filters.append(column.in_(value))

View File

@@ -4,11 +4,13 @@ import asyncio
from collections.abc import AsyncGenerator, Callable
from contextlib import AbstractAsyncContextManager, asynccontextmanager
from enum import Enum
from typing import Any, TypeVar
from typing import Any, TypeVar, cast
from sqlalchemy import text
from sqlalchemy import Table, delete, text, tuple_
from sqlalchemy.dialects.postgresql import insert as pg_insert
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import DeclarativeBase, QueryableAttribute
from sqlalchemy.orm.relationships import RelationshipProperty
from .exceptions import NotFoundError
@@ -20,6 +22,9 @@ __all__ = [
"create_db_dependency",
"get_transaction",
"lock_tables",
"m2m_add",
"m2m_remove",
"m2m_set",
"wait_for_row_change",
]
@@ -339,3 +344,140 @@ async def wait_for_row_change(
current = {col: getattr(instance, col) for col in watch_cols}
if current != initial:
return instance
def _m2m_prop(rel_attr: QueryableAttribute) -> RelationshipProperty: # type: ignore[type-arg]
"""Return the validated M2M RelationshipProperty for *rel_attr*.
Raises TypeError if *rel_attr* is not a Many-to-Many relationship.
"""
prop = rel_attr.property
if not isinstance(prop, RelationshipProperty) or prop.secondary is None:
raise TypeError(
f"m2m helpers require a Many-to-Many relationship attribute, "
f"got {rel_attr!r}. Use a relationship with a secondary table."
)
return prop
async def m2m_add(
session: AsyncSession,
instance: DeclarativeBase,
rel_attr: QueryableAttribute,
*related: DeclarativeBase,
ignore_conflicts: bool = False,
) -> None:
"""Insert rows into a Many-to-Many association table without loading the ORM collection.
Args:
session: DB async session.
instance: The "owner" side model instance (e.g. the ``A`` in ``A.b_list``).
rel_attr: The M2M relationship attribute on the model class (e.g. ``A.b_list``).
*related: One or more related instances to associate with ``instance``.
ignore_conflicts: When ``True``, silently skip rows that already exist
in the association table (``ON CONFLICT DO NOTHING``).
Raises:
TypeError: If ``rel_attr`` is not a Many-to-Many relationship.
"""
prop = _m2m_prop(rel_attr)
if not related:
return
secondary = cast(Table, prop.secondary)
assert secondary is not None # guaranteed by _m2m_prop
sync_pairs = prop.secondary_synchronize_pairs
assert sync_pairs is not None # set whenever secondary is set
# synchronize_pairs: [(parent_col, assoc_col), ...]
# secondary_synchronize_pairs: [(related_col, assoc_col), ...]
rows: list[dict[str, Any]] = []
for rel_instance in related:
row: dict[str, Any] = {}
for parent_col, assoc_col in prop.synchronize_pairs:
row[assoc_col.name] = getattr(instance, cast(str, parent_col.key))
for related_col, assoc_col in sync_pairs:
row[assoc_col.name] = getattr(rel_instance, cast(str, related_col.key))
rows.append(row)
stmt = pg_insert(secondary).values(rows)
if ignore_conflicts:
stmt = stmt.on_conflict_do_nothing()
await session.execute(stmt)
async def m2m_remove(
session: AsyncSession,
instance: DeclarativeBase,
rel_attr: QueryableAttribute,
*related: DeclarativeBase,
) -> None:
"""Remove rows from a Many-to-Many association table without loading the ORM collection.
Args:
session: DB async session.
instance: The "owner" side model instance (e.g. the ``A`` in ``A.b_list``).
rel_attr: The M2M relationship attribute on the model class (e.g. ``A.b_list``).
*related: One or more related instances to disassociate from ``instance``.
Raises:
TypeError: If ``rel_attr`` is not a Many-to-Many relationship.
"""
prop = _m2m_prop(rel_attr)
if not related:
return
secondary = cast(Table, prop.secondary)
assert secondary is not None # guaranteed by _m2m_prop
related_pairs = prop.secondary_synchronize_pairs
assert related_pairs is not None # set whenever secondary is set
parent_where = [
assoc_col == getattr(instance, cast(str, parent_col.key))
for parent_col, assoc_col in prop.synchronize_pairs
]
if len(related_pairs) == 1:
related_col, assoc_col = related_pairs[0]
related_values = [getattr(r, cast(str, related_col.key)) for r in related]
related_where = assoc_col.in_(related_values)
else:
assoc_cols = [ac for _, ac in related_pairs]
rel_cols = [rc for rc, _ in related_pairs]
related_values_t = [
tuple(getattr(r, cast(str, rc.key)) for rc in rel_cols) for r in related
]
related_where = tuple_(*assoc_cols).in_(related_values_t)
await session.execute(delete(secondary).where(*parent_where, related_where))
async def m2m_set(
session: AsyncSession,
instance: DeclarativeBase,
rel_attr: QueryableAttribute,
*related: DeclarativeBase,
) -> None:
"""Replace the entire Many-to-Many association set atomically.
Args:
session: DB async session.
instance: The "owner" side model instance (e.g. the ``A`` in ``A.b_list``).
rel_attr: The M2M relationship attribute on the model class (e.g. ``A.b_list``).
*related: The new complete set of related instances.
Raises:
TypeError: If ``rel_attr`` is not a Many-to-Many relationship.
"""
prop = _m2m_prop(rel_attr)
secondary = cast(Table, prop.secondary)
assert secondary is not None # guaranteed by _m2m_prop
parent_where = [
assoc_col == getattr(instance, cast(str, parent_col.key))
for parent_col, assoc_col in prop.synchronize_pairs
]
await session.execute(delete(secondary).where(*parent_where))
if related:
await m2m_add(session, instance, rel_attr, *related)

View File

@@ -2,12 +2,18 @@
from .enum import LoadStrategy
from .registry import Context, FixtureRegistry
from .utils import get_obj_by_attr, load_fixtures, load_fixtures_by_context
from .utils import (
get_field_by_attr,
get_obj_by_attr,
load_fixtures,
load_fixtures_by_context,
)
__all__ = [
"Context",
"FixtureRegistry",
"LoadStrategy",
"get_field_by_attr",
"get_obj_by_attr",
"load_fixtures",
"load_fixtures_by_context",

View File

@@ -250,6 +250,31 @@ def get_obj_by_attr(
) from None
def get_field_by_attr(
fixtures: Callable[[], Sequence[ModelType]],
attr_name: str,
value: Any,
*,
field: str = "id",
) -> Any:
"""Get a single field value from a fixture object matched by an attribute.
Args:
fixtures: A fixture function registered via ``@registry.register``
that returns a sequence of SQLAlchemy model instances.
attr_name: Name of the attribute to match against.
value: Value to match.
field: Attribute name to return from the matched object (default: ``"id"``).
Returns:
The value of ``field`` on the first matching model instance.
Raises:
StopIteration: If no matching object is found in the fixture group.
"""
return getattr(get_obj_by_attr(fixtures, attr_name, value), field)
async def load_fixtures(
session: AsyncSession,
registry: FixtureRegistry,

View File

@@ -1,11 +1,13 @@
"""Pytest plugin for using FixtureRegistry fixtures in tests."""
from collections.abc import Callable, Sequence
from typing import Any
from typing import Any, cast
import pytest
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import DeclarativeBase, selectinload
from sqlalchemy.orm.interfaces import ExecutableOption, ORMOption
from ..db import get_transaction
from ..fixtures import FixtureRegistry, LoadStrategy
@@ -112,7 +114,7 @@ def _create_fixture_function(
elif strategy == LoadStrategy.MERGE:
merged = await session.merge(instance)
loaded.append(merged)
elif strategy == LoadStrategy.SKIP_EXISTING:
elif strategy == LoadStrategy.SKIP_EXISTING: # pragma: no branch
pk = _get_primary_key(instance)
if pk is not None:
existing = await session.get(type(instance), pk)
@@ -125,6 +127,11 @@ def _create_fixture_function(
session.add(instance)
loaded.append(instance)
if loaded: # pragma: no branch
load_options = _relationship_load_options(type(loaded[0]))
if load_options:
return await _reload_with_relationships(session, loaded, load_options)
return loaded
# Update function signature to include dependencies
@@ -141,6 +148,54 @@ def _create_fixture_function(
return created_func
def _relationship_load_options(model: type[DeclarativeBase]) -> list[ExecutableOption]:
"""Build selectinload options for all direct relationships on a model."""
return [
selectinload(getattr(model, rel.key)) for rel in model.__mapper__.relationships
]
async def _reload_with_relationships(
session: AsyncSession,
instances: list[DeclarativeBase],
load_options: list[ExecutableOption],
) -> list[DeclarativeBase]:
"""Reload instances in a single bulk query with relationship eager-loading.
Uses one SELECT … WHERE pk IN (…) so selectinload can batch all relationship
queries — 1 + N_relationships round-trips regardless of how many instances
there are, instead of one session.get() per instance.
Preserves the original insertion order.
"""
model = type(instances[0])
mapper = model.__mapper__
pk_cols = mapper.primary_key
if len(pk_cols) == 1:
pk_attr = getattr(model, pk_cols[0].key)
pks = [getattr(inst, pk_cols[0].key) for inst in instances]
result = await session.execute(
select(model).where(pk_attr.in_(pks)).options(*load_options)
)
by_pk = {getattr(row, pk_cols[0].key): row for row in result.unique().scalars()}
return [by_pk[pk] for pk in pks]
# Composite PK: fall back to per-instance reload
reloaded: list[DeclarativeBase] = []
for instance in instances:
pk = _get_primary_key(instance)
refreshed = await session.get(
model,
pk,
options=cast(list[ORMOption], load_options),
populate_existing=True,
)
if refreshed is not None: # pragma: no branch
reloaded.append(refreshed)
return reloaded
def _get_primary_key(instance: DeclarativeBase) -> Any | None:
"""Get the primary key value of a model instance."""
mapper = instance.__class__.__mapper__

View File

@@ -163,6 +163,7 @@ class PaginatedResponse(BaseResponse, Generic[DataT]):
pagination_type: PaginationType | None = None
filter_attributes: dict[str, list[Any]] | None = None
search_columns: list[str] | None = None
order_columns: list[str] | None = None
_discriminated_union_cache: ClassVar[dict[Any, Any]] = {}

View File

@@ -15,13 +15,14 @@ ModelType = TypeVar("ModelType", bound=DeclarativeBase)
SchemaType = TypeVar("SchemaType", bound=BaseModel)
# CRUD type aliases
JoinType = list[tuple[type[DeclarativeBase], Any]]
JoinType = list[tuple[type[DeclarativeBase] | Any, Any]]
M2MFieldType = Mapping[str, QueryableAttribute[Any]]
OrderByClause = ColumnElement[Any] | QueryableAttribute[Any]
# Search / facet type aliases
# Search / facet / order type aliases
SearchFieldType = InstrumentedAttribute[Any] | tuple[InstrumentedAttribute[Any], ...]
FacetFieldType = SearchFieldType
OrderFieldType = SearchFieldType
# Dependency type aliases
SessionDependency = Callable[[], AsyncGenerator[AsyncSession, None]] | Any

View File

@@ -2,6 +2,7 @@
import os
import uuid
from enum import Enum
import pytest
from pydantic import BaseModel
@@ -12,6 +13,7 @@ from sqlalchemy import (
Column,
Date,
DateTime,
Enum as SAEnum,
ForeignKey,
Integer,
JSON,
@@ -139,6 +141,46 @@ class Post(Base):
tags: Mapped[list[Tag]] = relationship(secondary=post_tags)
class OrderStatus(int, Enum):
"""Integer-backed enum for order status."""
PENDING = 1
PROCESSING = 2
SHIPPED = 3
CANCELLED = 4
class Color(str, Enum):
"""String-backed enum for color."""
RED = "red"
GREEN = "green"
BLUE = "blue"
class Order(Base):
"""Test model with an IntEnum column (Enum(int, Enum)) and a raw Integer column."""
__tablename__ = "orders"
id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
name: Mapped[str] = mapped_column(String(100))
status: Mapped[OrderStatus] = mapped_column(SAEnum(OrderStatus))
priority: Mapped[int] = mapped_column(Integer)
color: Mapped[Color] = mapped_column(SAEnum(Color))
class Transfer(Base):
"""Test model with two FKs to the same table (users)."""
__tablename__ = "transfers"
id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
amount: Mapped[str] = mapped_column(String(50))
sender_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
receiver_id: Mapped[uuid.UUID] = mapped_column(ForeignKey("users.id"))
class Article(Base):
"""Test article model with ARRAY and JSON columns."""
@@ -300,6 +342,44 @@ class ArticleRead(PydanticBase):
labels: list[str]
class OrderCreate(BaseModel):
"""Schema for creating an order."""
id: uuid.UUID | None = None
name: str
status: OrderStatus
priority: int = 0
color: Color = Color.RED
class OrderRead(PydanticBase):
"""Schema for reading an order."""
id: uuid.UUID
name: str
status: OrderStatus
priority: int
color: Color
class TransferCreate(BaseModel):
"""Schema for creating a transfer."""
id: uuid.UUID | None = None
amount: str
sender_id: uuid.UUID
receiver_id: uuid.UUID
class TransferRead(PydanticBase):
"""Schema for reading a transfer."""
id: uuid.UUID
amount: str
OrderCrud = CrudFactory(Order)
TransferCrud = CrudFactory(Transfer)
ArticleCrud = CrudFactory(Article)
RoleCrud = CrudFactory(Role)
RoleCursorCrud = CrudFactory(Role, cursor_column=Role.id)

View File

@@ -38,6 +38,10 @@ from .conftest import (
Tag,
TagCreate,
TagCrud,
Transfer,
TransferCreate,
TransferCrud,
TransferRead,
User,
UserCreate,
UserCrud,
@@ -243,6 +247,60 @@ class TestResolveSearchColumns:
assert "username" not in result
class TestResolveOrderColumns:
"""Tests for _resolve_order_columns logic."""
def test_returns_none_when_no_order_fields(self):
"""Returns None when cls.order_fields is None and no order_fields passed."""
class AbstractCrud(AsyncCrud[User]):
pass
assert AbstractCrud._resolve_order_columns(None) is None
def test_returns_none_when_empty_order_fields_passed(self):
"""Returns None when an empty list is passed explicitly."""
crud = CrudFactory(User)
assert crud._resolve_order_columns([]) is None
def test_returns_keys_from_class_order_fields(self):
"""Returns sorted column keys from cls.order_fields when no override passed."""
crud = CrudFactory(User, order_fields=[User.username])
result = crud._resolve_order_columns(None)
assert result is not None
assert "username" in result
def test_order_fields_override_takes_priority(self):
"""Explicit order_fields override cls.order_fields."""
crud = CrudFactory(User, order_fields=[User.username])
result = crud._resolve_order_columns([User.email])
assert result is not None
assert "email" in result
assert "username" not in result
def test_returns_sorted_keys(self):
"""Keys are returned in sorted order."""
crud = CrudFactory(User, order_fields=[User.email, User.username])
result = crud._resolve_order_columns(None)
assert result is not None
assert result == sorted(result)
def test_relation_tuple_produces_dunder_key(self):
"""A (rel, column) tuple produces a 'rel__column' key."""
crud = CrudFactory(User, order_fields=[(User.role, Role.name)])
result = crud._resolve_order_columns(None)
assert result == ["role__name"]
def test_mixed_flat_and_relation_fields(self):
"""Flat and relation fields can be mixed; keys are sorted."""
crud = CrudFactory(User, order_fields=[User.username, (User.role, Role.name)])
result = crud._resolve_order_columns(None)
assert result is not None
assert "username" in result
assert "role__name" in result
assert result == sorted(result)
class TestDefaultLoadOptionsIntegration:
"""Integration tests for default_load_options with real DB queries."""
@@ -322,6 +380,43 @@ class TestDefaultLoadOptionsIntegration:
assert result.data[0].role is not None
assert result.data[0].role.name == "admin"
@pytest.mark.anyio
async def test_default_load_options_applied_to_create(
self, db_session: AsyncSession
):
"""default_load_options loads relationships after create()."""
UserWithDefaultLoad = CrudFactory(
User, default_load_options=[selectinload(User.role)]
)
role = await RoleCrud.create(db_session, RoleCreate(name="admin"))
user = await UserWithDefaultLoad.create(
db_session,
UserCreate(username="alice", email="alice@test.com", role_id=role.id),
)
assert user.role is not None
assert user.role.name == "admin"
@pytest.mark.anyio
async def test_default_load_options_applied_to_update(
self, db_session: AsyncSession
):
"""default_load_options loads relationships after update()."""
UserWithDefaultLoad = CrudFactory(
User, default_load_options=[selectinload(User.role)]
)
role = await RoleCrud.create(db_session, RoleCreate(name="admin"))
user = await UserCrud.create(
db_session,
UserCreate(username="alice", email="alice@test.com"),
)
updated = await UserWithDefaultLoad.update(
db_session,
UserUpdate(role_id=role.id),
filters=[User.id == user.id],
)
assert updated.role is not None
assert updated.role.name == "admin"
@pytest.mark.anyio
async def test_load_options_overrides_default_load_options(
self, db_session: AsyncSession
@@ -1282,6 +1377,128 @@ class TestCrudJoins:
assert users[0].username == "multi_join"
class TestCrudAliasedJoins:
"""Tests for CRUD operations with aliased joins (same table joined twice)."""
@pytest.mark.anyio
async def test_get_multi_with_aliased_joins(self, db_session: AsyncSession):
"""Aliased joins allow joining the same table twice."""
from sqlalchemy.orm import aliased
alice = await UserCrud.create(
db_session, UserCreate(username="alice", email="alice@test.com")
)
bob = await UserCrud.create(
db_session, UserCreate(username="bob", email="bob@test.com")
)
await TransferCrud.create(
db_session,
TransferCreate(amount="100", sender_id=alice.id, receiver_id=bob.id),
)
Sender = aliased(User)
Receiver = aliased(User)
results = await TransferCrud.get_multi(
db_session,
joins=[
(Sender, Transfer.sender_id == Sender.id),
(Receiver, Transfer.receiver_id == Receiver.id),
],
filters=[Sender.username == "alice", Receiver.username == "bob"],
)
assert len(results) == 1
assert results[0].amount == "100"
@pytest.mark.anyio
async def test_get_multi_aliased_no_match(self, db_session: AsyncSession):
"""Aliased joins correctly filter out non-matching rows."""
from sqlalchemy.orm import aliased
alice = await UserCrud.create(
db_session, UserCreate(username="alice", email="alice@test.com")
)
bob = await UserCrud.create(
db_session, UserCreate(username="bob", email="bob@test.com")
)
await TransferCrud.create(
db_session,
TransferCreate(amount="100", sender_id=alice.id, receiver_id=bob.id),
)
Sender = aliased(User)
Receiver = aliased(User)
# bob is receiver, not sender — should return nothing
results = await TransferCrud.get_multi(
db_session,
joins=[
(Sender, Transfer.sender_id == Sender.id),
(Receiver, Transfer.receiver_id == Receiver.id),
],
filters=[Sender.username == "bob", Receiver.username == "alice"],
)
assert len(results) == 0
@pytest.mark.anyio
async def test_paginate_with_aliased_joins(self, db_session: AsyncSession):
"""Aliased joins work with offset_paginate."""
from sqlalchemy.orm import aliased
alice = await UserCrud.create(
db_session, UserCreate(username="alice", email="alice@test.com")
)
bob = await UserCrud.create(
db_session, UserCreate(username="bob", email="bob@test.com")
)
await TransferCrud.create(
db_session,
TransferCreate(amount="50", sender_id=alice.id, receiver_id=bob.id),
)
await TransferCrud.create(
db_session,
TransferCreate(amount="75", sender_id=bob.id, receiver_id=alice.id),
)
Sender = aliased(User)
result = await TransferCrud.offset_paginate(
db_session,
joins=[(Sender, Transfer.sender_id == Sender.id)],
filters=[Sender.username == "alice"],
schema=TransferRead,
)
assert result.pagination.total_count == 1
assert result.data[0].amount == "50"
@pytest.mark.anyio
async def test_count_with_aliased_join(self, db_session: AsyncSession):
"""Aliased joins work with count."""
from sqlalchemy.orm import aliased
alice = await UserCrud.create(
db_session, UserCreate(username="alice", email="alice@test.com")
)
bob = await UserCrud.create(
db_session, UserCreate(username="bob", email="bob@test.com")
)
await TransferCrud.create(
db_session,
TransferCreate(amount="10", sender_id=alice.id, receiver_id=bob.id),
)
await TransferCrud.create(
db_session,
TransferCreate(amount="20", sender_id=alice.id, receiver_id=bob.id),
)
Sender = aliased(User)
count = await TransferCrud.count(
db_session,
joins=[(Sender, Transfer.sender_id == Sender.id)],
filters=[Sender.username == "alice"],
)
assert count == 2
class TestCrudFactoryM2M:
"""Tests for CrudFactory with m2m_fields parameter."""

View File

@@ -23,6 +23,12 @@ from .conftest import (
ArticleCreate,
ArticleCrud,
ArticleRead,
Color,
Order,
OrderCreate,
OrderCrud,
OrderRead,
OrderStatus,
Role,
RoleCreate,
RoleCrud,
@@ -971,7 +977,7 @@ class TestFilterBy:
@pytest.mark.anyio
async def test_bool_filter_false(self, db_session: AsyncSession):
"""filter_by with a boolean False value correctly filters rows."""
"""filter_by with a string 'false' value correctly filters rows."""
UserBoolCrud = CrudFactory(User, facet_fields=[User.is_active])
await UserCrud.create(
db_session, UserCreate(username="alice", email="a@test.com", is_active=True)
@@ -982,7 +988,7 @@ class TestFilterBy:
)
result = await UserBoolCrud.offset_paginate(
db_session, filter_by={"is_active": False}, schema=UserRead
db_session, filter_by={"is_active": "false"}, schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
@@ -991,7 +997,7 @@ class TestFilterBy:
@pytest.mark.anyio
async def test_bool_filter_true(self, db_session: AsyncSession):
"""filter_by with a boolean True value correctly filters rows."""
"""filter_by with a string 'true' value correctly filters rows."""
UserBoolCrud = CrudFactory(User, facet_fields=[User.is_active])
await UserCrud.create(
db_session, UserCreate(username="alice", email="a@test.com", is_active=True)
@@ -1002,7 +1008,7 @@ class TestFilterBy:
)
result = await UserBoolCrud.offset_paginate(
db_session, filter_by={"is_active": True}, schema=UserRead
db_session, filter_by={"is_active": "true"}, schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
@@ -1011,7 +1017,7 @@ class TestFilterBy:
@pytest.mark.anyio
async def test_bool_filter_list(self, db_session: AsyncSession):
"""filter_by with a list of booleans produces an IN clause."""
"""filter_by with a list of string booleans produces an IN clause."""
UserBoolCrud = CrudFactory(User, facet_fields=[User.is_active])
await UserCrud.create(
db_session, UserCreate(username="alice", email="a@test.com", is_active=True)
@@ -1022,12 +1028,41 @@ class TestFilterBy:
)
result = await UserBoolCrud.offset_paginate(
db_session, filter_by={"is_active": [True, False]}, schema=UserRead
db_session, filter_by={"is_active": ["true", "false"]}, schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 2
@pytest.mark.anyio
async def test_bool_filter_native_bool(self, db_session: AsyncSession):
"""filter_by with a native Python bool passes through coercion."""
UserBoolCrud = CrudFactory(User, facet_fields=[User.is_active])
await UserCrud.create(
db_session, UserCreate(username="alice", email="a@test.com", is_active=True)
)
result = await UserBoolCrud.offset_paginate(
db_session, filter_by={"is_active": True}, schema=UserRead
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 1
def test_bool_coerce_invalid_value(self):
"""_coerce_bool raises ValueError for non-bool, non-string values."""
from fastapi_toolsets.crud.search import _coerce_bool
with pytest.raises(ValueError, match="Cannot coerce"):
_coerce_bool(42)
def test_bool_coerce_invalid_string(self):
"""_coerce_bool raises ValueError for unrecognized string values."""
from fastapi_toolsets.crud.search import _coerce_bool
with pytest.raises(ValueError, match="Cannot coerce"):
_coerce_bool("maybe")
@pytest.mark.anyio
async def test_array_contains_single_value(self, db_session: AsyncSession):
"""filter_by on an ARRAY column with a scalar checks containment."""
@@ -1092,6 +1127,253 @@ class TestFilterBy:
assert "JSON" in exc_info.value.col_type
class TestFilterByIntEnum:
"""Tests for filter_by on columns typed as (int, Enum) / IntEnum."""
@pytest.mark.anyio
async def test_filter_by_intenum_member(self, db_session: AsyncSession):
"""filter_by with an IntEnum member value filters correctly."""
OrderFacetCrud = CrudFactory(Order, facet_fields=[Order.status])
await OrderCrud.create(
db_session, OrderCreate(name="order-1", status=OrderStatus.PENDING)
)
await OrderCrud.create(
db_session, OrderCreate(name="order-2", status=OrderStatus.SHIPPED)
)
await OrderCrud.create(
db_session, OrderCreate(name="order-3", status=OrderStatus.PENDING)
)
result = await OrderFacetCrud.offset_paginate(
db_session,
filter_by={"status": OrderStatus.PENDING},
schema=OrderRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 2
names = {o.name for o in result.data}
assert names == {"order-1", "order-3"}
@pytest.mark.anyio
async def test_filter_by_plain_int_value_raises(self, db_session: AsyncSession):
"""filter_by with a plain int on an IntEnum column raises KeyError — use name or member."""
OrderFacetCrud = CrudFactory(Order, facet_fields=[Order.status])
with pytest.raises(KeyError):
await OrderFacetCrud.offset_paginate(
db_session,
filter_by={"status": 1},
schema=OrderRead,
)
@pytest.mark.anyio
async def test_filter_by_intenum_list(self, db_session: AsyncSession):
"""filter_by with a list of IntEnum members produces an IN filter."""
OrderFacetCrud = CrudFactory(Order, facet_fields=[Order.status])
await OrderCrud.create(
db_session, OrderCreate(name="order-1", status=OrderStatus.PENDING)
)
await OrderCrud.create(
db_session, OrderCreate(name="order-2", status=OrderStatus.SHIPPED)
)
await OrderCrud.create(
db_session, OrderCreate(name="order-3", status=OrderStatus.CANCELLED)
)
result = await OrderFacetCrud.offset_paginate(
db_session,
filter_by={"status": [OrderStatus.PENDING, OrderStatus.SHIPPED]},
schema=OrderRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 2
names = {o.name for o in result.data}
assert names == {"order-1", "order-2"}
@pytest.mark.anyio
async def test_filter_by_plain_int_list_raises(self, db_session: AsyncSession):
"""filter_by with a list of plain ints on an IntEnum column raises KeyError — use names or members."""
OrderFacetCrud = CrudFactory(Order, facet_fields=[Order.status])
with pytest.raises(KeyError):
await OrderFacetCrud.offset_paginate(
db_session,
filter_by={"status": [1, 3]},
schema=OrderRead,
)
@pytest.mark.anyio
async def test_filter_by_intenum_name_string(self, db_session: AsyncSession):
"""filter_by with the enum member name as a string filters correctly."""
OrderFacetCrud = CrudFactory(Order, facet_fields=[Order.status])
await OrderCrud.create(
db_session, OrderCreate(name="order-1", status=OrderStatus.PENDING)
)
await OrderCrud.create(
db_session, OrderCreate(name="order-2", status=OrderStatus.SHIPPED)
)
result = await OrderFacetCrud.offset_paginate(
db_session,
filter_by={
"status": "PENDING"
}, # name as string, e.g. from HTTP query param
schema=OrderRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 1
assert result.data[0].name == "order-1"
@pytest.mark.anyio
async def test_filter_by_intenum_name_string_list(self, db_session: AsyncSession):
"""filter_by with a list of enum name strings produces an IN filter."""
OrderFacetCrud = CrudFactory(Order, facet_fields=[Order.status])
await OrderCrud.create(
db_session, OrderCreate(name="order-1", status=OrderStatus.PENDING)
)
await OrderCrud.create(
db_session, OrderCreate(name="order-2", status=OrderStatus.SHIPPED)
)
await OrderCrud.create(
db_session, OrderCreate(name="order-3", status=OrderStatus.CANCELLED)
)
result = await OrderFacetCrud.offset_paginate(
db_session,
filter_by={"status": ["PENDING", "SHIPPED"]},
schema=OrderRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 2
names = {o.name for o in result.data}
assert names == {"order-1", "order-2"}
class TestFilterByStrEnum:
"""Tests for filter_by on columns typed as (str, Enum) / StrEnum (lines 364-367)."""
@pytest.mark.anyio
async def test_filter_by_strenum_member(self, db_session: AsyncSession):
"""filter_by with a StrEnum member on a string Enum column filters correctly."""
OrderColorCrud = CrudFactory(Order, facet_fields=[Order.color])
await OrderCrud.create(
db_session,
OrderCreate(name="red-order", status=OrderStatus.PENDING, color=Color.RED),
)
await OrderCrud.create(
db_session,
OrderCreate(
name="blue-order", status=OrderStatus.PENDING, color=Color.BLUE
),
)
result = await OrderColorCrud.offset_paginate(
db_session,
filter_by={"color": Color.RED},
schema=OrderRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 1
assert result.data[0].name == "red-order"
@pytest.mark.anyio
async def test_filter_by_strenum_list(self, db_session: AsyncSession):
"""filter_by with a list of StrEnum members produces an IN filter."""
OrderColorCrud = CrudFactory(Order, facet_fields=[Order.color])
await OrderCrud.create(
db_session,
OrderCreate(name="red-order", status=OrderStatus.PENDING, color=Color.RED),
)
await OrderCrud.create(
db_session,
OrderCreate(
name="green-order", status=OrderStatus.PENDING, color=Color.GREEN
),
)
await OrderCrud.create(
db_session,
OrderCreate(
name="blue-order", status=OrderStatus.PENDING, color=Color.BLUE
),
)
result = await OrderColorCrud.offset_paginate(
db_session,
filter_by={"color": [Color.RED, Color.BLUE]},
schema=OrderRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 2
names = {o.name for o in result.data}
assert names == {"red-order", "blue-order"}
class TestFilterByIntegerColumn:
"""Tests for filter_by on plain Integer columns with IntEnum values."""
@pytest.mark.anyio
async def test_filter_by_integer_column_with_intenum_member(
self, db_session: AsyncSession
):
"""filter_by with an IntEnum member on an Integer column works correctly."""
OrderPriorityCrud = CrudFactory(Order, facet_fields=[Order.priority])
await OrderCrud.create(
db_session,
OrderCreate(
name="order-1", status=OrderStatus.PENDING, priority=OrderStatus.PENDING
),
)
await OrderCrud.create(
db_session,
OrderCreate(
name="order-2", status=OrderStatus.SHIPPED, priority=OrderStatus.SHIPPED
),
)
result = await OrderPriorityCrud.offset_paginate(
db_session,
filter_by={
"priority": OrderStatus.PENDING
}, # IntEnum member on Integer col
schema=OrderRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 1
assert result.data[0].name == "order-1"
@pytest.mark.anyio
async def test_filter_by_integer_column_with_plain_int(
self, db_session: AsyncSession
):
"""filter_by with a plain int on an Integer column works correctly."""
OrderPriorityCrud = CrudFactory(Order, facet_fields=[Order.priority])
await OrderCrud.create(
db_session,
OrderCreate(name="order-1", status=OrderStatus.PENDING, priority=1),
)
await OrderCrud.create(
db_session,
OrderCreate(name="order-2", status=OrderStatus.SHIPPED, priority=3),
)
result = await OrderPriorityCrud.offset_paginate(
db_session,
filter_by={"priority": 1},
schema=OrderRead,
)
assert isinstance(result.pagination, OffsetPagination)
assert result.pagination.total_count == 1
assert result.data[0].name == "order-1"
class TestFilterParamsViaConsolidated:
"""Tests for filter params via consolidated offset_paginate_params()."""
@@ -1487,6 +1769,111 @@ class TestSearchColumns:
assert result.data[0].username == "bob"
class TestOrderColumns:
"""Tests for order_columns in paginated responses."""
@pytest.mark.anyio
async def test_order_columns_returned_in_offset_paginate(
self, db_session: AsyncSession
):
"""offset_paginate response includes order_columns."""
UserSortCrud = CrudFactory(User, order_fields=[User.username, User.email])
await UserCrud.create(
db_session, UserCreate(username="alice", email="a@test.com")
)
result = await UserSortCrud.offset_paginate(db_session, schema=UserRead)
assert result.order_columns is not None
assert "username" in result.order_columns
assert "email" in result.order_columns
@pytest.mark.anyio
async def test_order_columns_returned_in_cursor_paginate(
self, db_session: AsyncSession
):
"""cursor_paginate response includes order_columns."""
UserSortCursorCrud = CrudFactory(
User,
cursor_column=User.id,
order_fields=[User.username, User.email],
)
await UserCrud.create(
db_session, UserCreate(username="alice", email="a@test.com")
)
result = await UserSortCursorCrud.cursor_paginate(db_session, schema=UserRead)
assert result.order_columns is not None
assert "username" in result.order_columns
assert "email" in result.order_columns
@pytest.mark.anyio
async def test_order_columns_none_when_no_order_fields(
self, db_session: AsyncSession
):
"""order_columns is None when no order_fields are configured."""
result = await UserCrud.offset_paginate(db_session, schema=UserRead)
assert result.order_columns is None
@pytest.mark.anyio
async def test_order_columns_override_in_offset_paginate(
self, db_session: AsyncSession
):
"""order_fields override in offset_paginate is reflected in order_columns."""
await UserCrud.create(
db_session, UserCreate(username="alice", email="a@test.com")
)
result = await UserCrud.offset_paginate(
db_session, order_fields=[User.email], schema=UserRead
)
assert result.order_columns == ["email"]
@pytest.mark.anyio
async def test_order_columns_override_in_cursor_paginate(
self, db_session: AsyncSession
):
"""order_fields override in cursor_paginate is reflected in order_columns."""
UserCursorCrud = CrudFactory(User, cursor_column=User.id)
await UserCrud.create(
db_session, UserCreate(username="alice", email="a@test.com")
)
result = await UserCursorCrud.cursor_paginate(
db_session, order_fields=[User.username], schema=UserRead
)
assert result.order_columns == ["username"]
@pytest.mark.anyio
async def test_order_columns_are_sorted_alphabetically(
self, db_session: AsyncSession
):
"""order_columns keys are returned in alphabetical order."""
UserSortCrud = CrudFactory(User, order_fields=[User.email, User.username])
await UserCrud.create(
db_session, UserCreate(username="alice", email="a@test.com")
)
result = await UserSortCrud.offset_paginate(db_session, schema=UserRead)
assert result.order_columns is not None
assert result.order_columns == sorted(result.order_columns)
@pytest.mark.anyio
async def test_relation_order_field_in_order_columns(
self, db_session: AsyncSession
):
"""A relation tuple order field produces 'rel__column' key in order_columns."""
UserSortCrud = CrudFactory(User, order_fields=[(User.role, Role.name)])
result = await UserSortCrud.offset_paginate(db_session, schema=UserRead)
assert result.order_columns == ["role__name"]
class TestOrderParamsViaConsolidated:
"""Tests for order params via consolidated offset_paginate_params()."""
@@ -1641,6 +2028,92 @@ class TestOrderParamsViaConsolidated:
assert result.data[0].username == "alice"
assert result.data[1].username == "charlie"
def test_relation_order_field_key_in_enum(self):
"""A relation tuple field produces a 'rel__column' key in the order_by enum."""
UserOrderCrud = CrudFactory(User, order_fields=[(User.role, Role.name)])
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
sig = inspect.signature(dep)
description = sig.parameters["order_by"].default.description
assert "role__name" in description
@pytest.mark.anyio
async def test_relation_order_field_produces_order_joins(self):
"""Selecting a relation order field emits order_by and order_joins."""
UserOrderCrud = CrudFactory(User, order_fields=[(User.role, Role.name)])
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
result = await dep(
page=1, items_per_page=20, order_by="role__name", order="asc"
)
assert "order_by" in result
assert "order_joins" in result
assert result["order_joins"] == [User.role]
@pytest.mark.anyio
async def test_relation_order_integrates_with_offset_paginate(
self, db_session: AsyncSession
):
"""Relation order field joins the related table and sorts correctly."""
UserOrderCrud = CrudFactory(User, order_fields=[(User.role, Role.name)])
role_b = await RoleCrud.create(db_session, RoleCreate(name="beta"))
role_a = await RoleCrud.create(db_session, RoleCreate(name="alpha"))
await UserCrud.create(
db_session,
UserCreate(username="u1", email="u1@test.com", role_id=role_b.id),
)
await UserCrud.create(
db_session,
UserCreate(username="u2", email="u2@test.com", role_id=role_a.id),
)
await UserCrud.create(
db_session, UserCreate(username="u3", email="u3@test.com")
)
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
params = await dep(
page=1, items_per_page=20, order_by="role__name", order="asc"
)
result = await UserOrderCrud.offset_paginate(
db_session, **params, schema=UserRead
)
usernames = [u.username for u in result.data]
# u2 (alpha) before u1 (beta); u3 (no role, NULL) comes last or first depending on DB
assert usernames.index("u2") < usernames.index("u1")
@pytest.mark.anyio
async def test_relation_order_integrates_with_cursor_paginate(
self, db_session: AsyncSession
):
"""Relation order field works with cursor_paginate (order_joins applied)."""
UserOrderCrud = CrudFactory(
User,
order_fields=[(User.role, Role.name)],
cursor_column=User.id,
)
role_b = await RoleCrud.create(db_session, RoleCreate(name="zeta"))
role_a = await RoleCrud.create(db_session, RoleCreate(name="alpha"))
await UserCrud.create(
db_session,
UserCreate(username="cx1", email="cx1@test.com", role_id=role_b.id),
)
await UserCrud.create(
db_session,
UserCreate(username="cx2", email="cx2@test.com", role_id=role_a.id),
)
dep = UserOrderCrud.cursor_paginate_params(search=False, filter=False)
params = await dep(
cursor=None, items_per_page=20, order_by="role__name", order="asc"
)
result = await UserOrderCrud.cursor_paginate(
db_session, **params, schema=UserRead
)
assert result.data is not None
assert len(result.data) == 2
class TestOffsetPaginateParamsSchema:
"""Tests for AsyncCrud.offset_paginate_params()."""

View File

@@ -4,10 +4,26 @@ import asyncio
import uuid
import pytest
from sqlalchemy import text
from sqlalchemy import (
Column,
ForeignKey,
ForeignKeyConstraint,
String,
Table,
Uuid,
select,
text,
)
from sqlalchemy.engine import make_url
from sqlalchemy.exc import IntegrityError
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import (
DeclarativeBase,
Mapped,
mapped_column,
relationship,
selectinload,
)
from fastapi_toolsets.db import (
LockMode,
@@ -17,12 +33,15 @@ from fastapi_toolsets.db import (
create_db_dependency,
get_transaction,
lock_tables,
m2m_add,
m2m_remove,
m2m_set,
wait_for_row_change,
)
from fastapi_toolsets.exceptions import NotFoundError
from fastapi_toolsets.pytest import create_db_session
from .conftest import DATABASE_URL, Base, Role, RoleCrud, User, UserCrud
from .conftest import DATABASE_URL, Base, Post, Role, RoleCrud, Tag, User, UserCrud
class TestCreateDbDependency:
@@ -81,6 +100,21 @@ class TestCreateDbDependency:
await engine.dispose()
@pytest.mark.anyio
async def test_no_commit_when_not_in_transaction(self):
"""Dependency skips commit if the session is no longer in a transaction on exit."""
engine = create_async_engine(DATABASE_URL, echo=False)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
get_db = create_db_dependency(session_factory)
async for session in get_db():
# Manually commit — session exits the transaction
await session.commit()
assert not session.in_transaction()
# The dependency's post-yield path must not call commit again (no error)
await engine.dispose()
@pytest.mark.anyio
async def test_update_after_lock_tables_is_persisted(self):
"""Changes made after lock_tables exits (before endpoint returns) are committed.
@@ -480,3 +514,417 @@ class TestCleanupTables:
async with create_db_session(DATABASE_URL, Base, drop_tables=True) as session:
# Should not raise
await cleanup_tables(session, EmptyBase)
class TestM2MAdd:
"""Tests for m2m_add helper."""
@pytest.mark.anyio
async def test_adds_single_related(self, db_session: AsyncSession):
"""Associates one related instance via the secondary table."""
user = User(username="m2m_author", email="m2m@test.com")
db_session.add(user)
await db_session.flush()
post = Post(title="Post A", author_id=user.id)
tag = Tag(name="python")
db_session.add_all([post, tag])
await db_session.flush()
async with get_transaction(db_session):
await m2m_add(db_session, post, Post.tags, tag)
result = await db_session.execute(
select(Post).where(Post.id == post.id).options(selectinload(Post.tags))
)
loaded = result.scalar_one()
assert len(loaded.tags) == 1
assert loaded.tags[0].id == tag.id
@pytest.mark.anyio
async def test_adds_multiple_related(self, db_session: AsyncSession):
"""Associates multiple related instances in a single call."""
user = User(username="m2m_author2", email="m2m2@test.com")
db_session.add(user)
await db_session.flush()
post = Post(title="Post B", author_id=user.id)
tag1 = Tag(name="web")
tag2 = Tag(name="api")
tag3 = Tag(name="async")
db_session.add_all([post, tag1, tag2, tag3])
await db_session.flush()
async with get_transaction(db_session):
await m2m_add(db_session, post, Post.tags, tag1, tag2, tag3)
result = await db_session.execute(
select(Post).where(Post.id == post.id).options(selectinload(Post.tags))
)
loaded = result.scalar_one()
assert {t.id for t in loaded.tags} == {tag1.id, tag2.id, tag3.id}
@pytest.mark.anyio
async def test_noop_for_empty_related(self, db_session: AsyncSession):
"""Calling with no related instances is a no-op."""
user = User(username="m2m_author3", email="m2m3@test.com")
db_session.add(user)
await db_session.flush()
post = Post(title="Post C", author_id=user.id)
db_session.add(post)
await db_session.flush()
async with get_transaction(db_session):
await m2m_add(db_session, post, Post.tags) # no related instances
result = await db_session.execute(
select(Post).where(Post.id == post.id).options(selectinload(Post.tags))
)
loaded = result.scalar_one()
assert loaded.tags == []
@pytest.mark.anyio
async def test_ignore_conflicts_true(self, db_session: AsyncSession):
"""Duplicate inserts are silently skipped when ignore_conflicts=True."""
user = User(username="m2m_author4", email="m2m4@test.com")
db_session.add(user)
await db_session.flush()
post = Post(title="Post D", author_id=user.id)
tag = Tag(name="duplicate_tag")
db_session.add_all([post, tag])
await db_session.flush()
async with get_transaction(db_session):
await m2m_add(db_session, post, Post.tags, tag)
# Second call with ignore_conflicts=True must not raise
async with get_transaction(db_session):
await m2m_add(db_session, post, Post.tags, tag, ignore_conflicts=True)
result = await db_session.execute(
select(Post).where(Post.id == post.id).options(selectinload(Post.tags))
)
loaded = result.scalar_one()
assert len(loaded.tags) == 1
@pytest.mark.anyio
async def test_ignore_conflicts_false_raises(self, db_session: AsyncSession):
"""Duplicate inserts raise IntegrityError when ignore_conflicts=False (default)."""
user = User(username="m2m_author5", email="m2m5@test.com")
db_session.add(user)
await db_session.flush()
post = Post(title="Post E", author_id=user.id)
tag = Tag(name="conflict_tag")
db_session.add_all([post, tag])
await db_session.flush()
async with get_transaction(db_session):
await m2m_add(db_session, post, Post.tags, tag)
with pytest.raises(IntegrityError):
async with get_transaction(db_session):
await m2m_add(db_session, post, Post.tags, tag)
@pytest.mark.anyio
async def test_non_m2m_raises_type_error(self, db_session: AsyncSession):
"""Passing a non-M2M relationship attribute raises TypeError."""
user = User(username="m2m_author6", email="m2m6@test.com")
db_session.add(user)
await db_session.flush()
role = Role(name="type_err_role")
db_session.add(role)
await db_session.flush()
with pytest.raises(TypeError, match="Many-to-Many"):
await m2m_add(db_session, user, User.role, role)
@pytest.mark.anyio
async def test_works_inside_lock_tables(self, db_session: AsyncSession):
"""m2m_add works correctly inside a lock_tables nested transaction."""
user = User(username="m2m_lock_author", email="m2m_lock@test.com")
db_session.add(user)
await db_session.flush()
async with lock_tables(db_session, [Tag]):
tag = Tag(name="locked_tag")
db_session.add(tag)
await db_session.flush()
post = Post(title="Post Lock", author_id=user.id)
db_session.add(post)
await db_session.flush()
await m2m_add(db_session, post, Post.tags, tag)
result = await db_session.execute(
select(Post).where(Post.id == post.id).options(selectinload(Post.tags))
)
loaded = result.scalar_one()
assert len(loaded.tags) == 1
assert loaded.tags[0].name == "locked_tag"
class _LocalBase(DeclarativeBase):
pass
_comp_assoc = Table(
"_comp_assoc",
_LocalBase.metadata,
Column("owner_id", Uuid, ForeignKey("_comp_owners.id"), primary_key=True),
Column("item_group", String(50), primary_key=True),
Column("item_code", String(50), primary_key=True),
ForeignKeyConstraint(
["item_group", "item_code"],
["_comp_items.group_id", "_comp_items.item_code"],
),
)
class _CompOwner(_LocalBase):
__tablename__ = "_comp_owners"
id: Mapped[uuid.UUID] = mapped_column(Uuid, primary_key=True, default=uuid.uuid4)
items: Mapped[list["_CompItem"]] = relationship(secondary=_comp_assoc)
class _CompItem(_LocalBase):
__tablename__ = "_comp_items"
group_id: Mapped[str] = mapped_column(String(50), primary_key=True)
item_code: Mapped[str] = mapped_column(String(50), primary_key=True)
class TestM2MRemove:
"""Tests for m2m_remove helper."""
async def _setup(
self, session: AsyncSession, username: str, email: str, *tag_names: str
):
"""Create a user, post, and tags; associate all tags with the post."""
user = User(username=username, email=email)
session.add(user)
await session.flush()
post = Post(title=f"Post {username}", author_id=user.id)
tags = [Tag(name=n) for n in tag_names]
session.add(post)
session.add_all(tags)
await session.flush()
async with get_transaction(session):
await m2m_add(session, post, Post.tags, *tags)
return post, tags
async def _load_tags(self, session: AsyncSession, post: Post) -> list[Tag]:
result = await session.execute(
select(Post).where(Post.id == post.id).options(selectinload(Post.tags))
)
return result.scalar_one().tags
@pytest.mark.anyio
async def test_removes_single(self, db_session: AsyncSession):
"""Removes one association, leaving others intact."""
post, (tag1, tag2) = await self._setup(
db_session, "rm_author1", "rm1@test.com", "tag_rm_a", "tag_rm_b"
)
async with get_transaction(db_session):
await m2m_remove(db_session, post, Post.tags, tag1)
remaining = await self._load_tags(db_session, post)
assert len(remaining) == 1
assert remaining[0].id == tag2.id
@pytest.mark.anyio
async def test_removes_multiple(self, db_session: AsyncSession):
"""Removes multiple associations in one call."""
post, (tag1, tag2, tag3) = await self._setup(
db_session, "rm_author2", "rm2@test.com", "tag_rm_c", "tag_rm_d", "tag_rm_e"
)
async with get_transaction(db_session):
await m2m_remove(db_session, post, Post.tags, tag1, tag3)
remaining = await self._load_tags(db_session, post)
assert len(remaining) == 1
assert remaining[0].id == tag2.id
@pytest.mark.anyio
async def test_noop_for_empty_related(self, db_session: AsyncSession):
"""Calling with no related instances is a no-op."""
post, (tag,) = await self._setup(
db_session, "rm_author3", "rm3@test.com", "tag_rm_f"
)
async with get_transaction(db_session):
await m2m_remove(db_session, post, Post.tags)
remaining = await self._load_tags(db_session, post)
assert len(remaining) == 1
@pytest.mark.anyio
async def test_idempotent_for_missing_association(self, db_session: AsyncSession):
"""Removing a non-existent association does not raise."""
post, (tag1,) = await self._setup(
db_session, "rm_author4", "rm4@test.com", "tag_rm_g"
)
tag2 = Tag(name="tag_rm_h")
db_session.add(tag2)
await db_session.flush()
# tag2 was never associated — should not raise
async with get_transaction(db_session):
await m2m_remove(db_session, post, Post.tags, tag2)
remaining = await self._load_tags(db_session, post)
assert len(remaining) == 1
@pytest.mark.anyio
async def test_non_m2m_raises_type_error(self, db_session: AsyncSession):
"""Passing a non-M2M relationship attribute raises TypeError."""
user = User(username="rm_author5", email="rm5@test.com")
db_session.add(user)
await db_session.flush()
role = Role(name="rm_type_err_role")
db_session.add(role)
await db_session.flush()
with pytest.raises(TypeError, match="Many-to-Many"):
await m2m_remove(db_session, user, User.role, role)
@pytest.mark.anyio
async def test_removes_composite_pk_related(self):
"""Composite-PK branch: DELETE uses tuple IN when related side has multi-col PK."""
engine = create_async_engine(DATABASE_URL, echo=False)
async with engine.begin() as conn:
await conn.run_sync(_LocalBase.metadata.create_all)
session_factory = async_sessionmaker(engine, expire_on_commit=False)
try:
async with session_factory() as session:
owner = _CompOwner()
item1 = _CompItem(group_id="g1", item_code="c1")
item2 = _CompItem(group_id="g1", item_code="c2")
session.add_all([owner, item1, item2])
await session.flush()
async with get_transaction(session):
await m2m_add(session, owner, _CompOwner.items, item1, item2)
async with get_transaction(session):
await m2m_remove(session, owner, _CompOwner.items, item1)
await session.commit()
async with session_factory() as verify:
from sqlalchemy import select
from sqlalchemy.orm import selectinload
result = await verify.execute(
select(_CompOwner)
.where(_CompOwner.id == owner.id)
.options(selectinload(_CompOwner.items))
)
loaded = result.scalar_one()
assert len(loaded.items) == 1
assert (loaded.items[0].group_id, loaded.items[0].item_code) == (
"g1",
"c2",
)
finally:
async with engine.begin() as conn:
await conn.run_sync(_LocalBase.metadata.drop_all)
await engine.dispose()
class TestM2MSet:
"""Tests for m2m_set helper."""
async def _load_tags(self, session: AsyncSession, post: Post) -> list[Tag]:
result = await session.execute(
select(Post).where(Post.id == post.id).options(selectinload(Post.tags))
)
return result.scalar_one().tags
@pytest.mark.anyio
async def test_replaces_existing_set(self, db_session: AsyncSession):
"""Replaces the full association set atomically."""
user = User(username="set_author1", email="set1@test.com")
db_session.add(user)
await db_session.flush()
post = Post(title="Post Set A", author_id=user.id)
tag1 = Tag(name="tag_set_a")
tag2 = Tag(name="tag_set_b")
tag3 = Tag(name="tag_set_c")
db_session.add_all([post, tag1, tag2, tag3])
await db_session.flush()
async with get_transaction(db_session):
await m2m_add(db_session, post, Post.tags, tag1, tag2)
async with get_transaction(db_session):
await m2m_set(db_session, post, Post.tags, tag3)
remaining = await self._load_tags(db_session, post)
assert len(remaining) == 1
assert remaining[0].id == tag3.id
@pytest.mark.anyio
async def test_clears_all_when_no_related(self, db_session: AsyncSession):
"""Passing no related instances clears all associations."""
user = User(username="set_author2", email="set2@test.com")
db_session.add(user)
await db_session.flush()
post = Post(title="Post Set B", author_id=user.id)
tag = Tag(name="tag_set_d")
db_session.add_all([post, tag])
await db_session.flush()
async with get_transaction(db_session):
await m2m_add(db_session, post, Post.tags, tag)
async with get_transaction(db_session):
await m2m_set(db_session, post, Post.tags)
remaining = await self._load_tags(db_session, post)
assert remaining == []
@pytest.mark.anyio
async def test_set_on_empty_then_populate(self, db_session: AsyncSession):
"""m2m_set works on a post with no existing associations."""
user = User(username="set_author3", email="set3@test.com")
db_session.add(user)
await db_session.flush()
post = Post(title="Post Set C", author_id=user.id)
tag1 = Tag(name="tag_set_e")
tag2 = Tag(name="tag_set_f")
db_session.add_all([post, tag1, tag2])
await db_session.flush()
async with get_transaction(db_session):
await m2m_set(db_session, post, Post.tags, tag1, tag2)
remaining = await self._load_tags(db_session, post)
assert {t.id for t in remaining} == {tag1.id, tag2.id}
@pytest.mark.anyio
async def test_non_m2m_raises_type_error(self, db_session: AsyncSession):
"""Passing a non-M2M relationship attribute raises TypeError."""
user = User(username="set_author4", email="set4@test.com")
db_session.add(user)
await db_session.flush()
role = Role(name="set_type_err_role")
db_session.add(role)
await db_session.flush()
with pytest.raises(TypeError, match="Many-to-Many"):
await m2m_set(db_session, user, User.role, role)

View File

@@ -10,6 +10,7 @@ from fastapi_toolsets.fixtures import (
Context,
FixtureRegistry,
LoadStrategy,
get_field_by_attr,
get_obj_by_attr,
load_fixtures,
load_fixtures_by_context,
@@ -951,6 +952,41 @@ class TestGetObjByAttr:
get_obj_by_attr(self.roles, "id", "not-a-uuid")
class TestGetFieldByAttr:
"""Tests for get_field_by_attr helper function."""
def setup_method(self):
self.registry = FixtureRegistry()
self.role_id_1 = uuid.uuid4()
self.role_id_2 = uuid.uuid4()
role_id_1 = self.role_id_1
role_id_2 = self.role_id_2
@self.registry.register
def roles() -> list[Role]:
return [
Role(id=role_id_1, name="admin"),
Role(id=role_id_2, name="user"),
]
self.roles = roles
def test_returns_id_by_default(self):
"""Returns the id field when no field is specified."""
result = get_field_by_attr(self.roles, "name", "admin")
assert result == self.role_id_1
def test_returns_specified_field(self):
"""Returns the requested field instead of id."""
result = get_field_by_attr(self.roles, "id", self.role_id_2, field="name")
assert result == "user"
def test_no_match_raises_stop_iteration(self):
"""Propagates StopIteration from get_obj_by_attr when no match found."""
with pytest.raises(StopIteration, match="No object with name=missing"):
get_field_by_attr(self.roles, "name", "missing")
class TestGetPrimaryKey:
"""Unit tests for the _get_primary_key helper (composite PK paths)."""

View File

@@ -1,17 +1,18 @@
"""Tests for fastapi_toolsets.pytest module."""
import uuid
from typing import cast
import pytest
from fastapi import Depends, FastAPI
from httpx import AsyncClient
from sqlalchemy import select, text
from sqlalchemy import ForeignKey, String, select, text
from sqlalchemy.engine import make_url
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from sqlalchemy.orm import selectinload
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
from fastapi_toolsets.db import get_transaction
from fastapi_toolsets.fixtures import Context, FixtureRegistry
from fastapi_toolsets.fixtures import Context, FixtureRegistry, LoadStrategy
from fastapi_toolsets.pytest import (
create_async_client,
create_db_session,
@@ -19,9 +20,23 @@ from fastapi_toolsets.pytest import (
register_fixtures,
worker_database_url,
)
from fastapi_toolsets.pytest.plugin import (
_get_primary_key,
_relationship_load_options,
_reload_with_relationships,
)
from fastapi_toolsets.pytest.utils import _get_xdist_worker
from .conftest import DATABASE_URL, Base, Role, RoleCrud, User, UserCrud
from .conftest import (
DATABASE_URL,
Base,
IntRole,
Permission,
Role,
RoleCrud,
User,
UserCrud,
)
test_registry = FixtureRegistry()
@@ -136,14 +151,8 @@ class TestGeneratedFixtures:
async def test_fixture_relationships_work(
self, db_session: AsyncSession, fixture_users: list[User]
):
"""Loaded fixtures have working relationships."""
# Load user with role relationship
user = await UserCrud.get(
db_session,
[User.id == USER_ADMIN_ID],
load_options=[selectinload(User.role)],
)
"""Loaded fixtures have working relationships directly accessible."""
user = next(u for u in fixture_users if u.id == USER_ADMIN_ID)
assert user.role is not None
assert user.role.name == "plugin_admin"
@@ -177,6 +186,15 @@ class TestGeneratedFixtures:
assert users[0].username == "plugin_admin"
assert users[1].username == "plugin_user"
@pytest.mark.anyio
async def test_fixture_auto_loads_relationships(
self, db_session: AsyncSession, fixture_users: list[User]
):
"""Fixtures automatically eager-load all direct relationships."""
user = next(u for u in fixture_users if u.username == "plugin_admin")
assert user.role is not None
assert user.role.name == "plugin_admin"
@pytest.mark.anyio
async def test_multiple_fixtures_in_same_test(
self,
@@ -516,3 +534,192 @@ class TestCreateWorkerDatabase:
)
assert result.scalar() is None
await engine.dispose()
class _LocalBase(DeclarativeBase):
pass
class _Group(_LocalBase):
__tablename__ = "_test_groups"
id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
name: Mapped[str] = mapped_column(String(50))
class _CompositeItem(_LocalBase):
"""Model with composite PK and a relationship — exercises the fallback path."""
__tablename__ = "_test_composite_items"
group_id: Mapped[uuid.UUID] = mapped_column(
ForeignKey("_test_groups.id"), primary_key=True
)
item_code: Mapped[str] = mapped_column(String(50), primary_key=True)
group: Mapped["_Group"] = relationship()
class TestGetPrimaryKey:
"""Unit tests for _get_primary_key — no DB needed."""
def test_single_pk_returns_value(self):
rid = uuid.UUID("00000000-0000-0000-0000-000000000001")
role = Role(id=rid, name="x")
assert _get_primary_key(role) == rid
def test_composite_pk_all_set_returns_tuple(self):
perm = Permission(subject="posts", action="read")
assert _get_primary_key(perm) == ("posts", "read")
def test_composite_pk_partial_none_returns_none(self):
perm = Permission(subject=None, action="read")
assert _get_primary_key(perm) is None
def test_composite_pk_all_none_returns_none(self):
perm = Permission(subject=None, action=None)
assert _get_primary_key(perm) is None
class TestRelationshipLoadOptions:
"""Unit tests for _relationship_load_options — no DB needed."""
def test_empty_for_model_with_no_relationships(self):
assert _relationship_load_options(IntRole) == []
def test_returns_options_for_model_with_relationships(self):
opts = _relationship_load_options(User)
assert len(opts) >= 1
class TestFixtureStrategies:
"""Integration tests covering INSERT, SKIP_EXISTING, empty fixture, no-rels model."""
@pytest.mark.anyio
async def test_empty_fixture_returns_empty_list(self, db_session: AsyncSession):
"""Fixture function returning [] produces an empty list."""
registry = FixtureRegistry()
@registry.register()
def empty() -> list[Role]:
return []
local_ns: dict = {}
register_fixtures(registry, local_ns, session_fixture="db_session")
inner = local_ns["fixture_empty"].__wrapped__ # type: ignore[attr-defined]
result = await inner(db_session=db_session)
assert result == []
@pytest.mark.anyio
async def test_insert_strategy_no_relationships(self, db_session: AsyncSession):
"""INSERT strategy adds instances; model with no rels skips reload (line 135)."""
registry = FixtureRegistry()
@registry.register()
def int_roles() -> list[IntRole]:
return [IntRole(name="insert_role")]
local_ns: dict = {}
register_fixtures(
registry,
local_ns,
session_fixture="db_session",
strategy=LoadStrategy.INSERT,
)
inner = local_ns["fixture_int_roles"].__wrapped__ # type: ignore[attr-defined]
result = await inner(db_session=db_session)
assert len(result) == 1
assert result[0].name == "insert_role"
@pytest.mark.anyio
async def test_skip_existing_inserts_new_record(self, db_session: AsyncSession):
"""SKIP_EXISTING inserts when the record does not yet exist."""
registry = FixtureRegistry()
role_id = uuid.uuid4()
@registry.register()
def new_roles() -> list[Role]:
return [Role(id=role_id, name="skip_new")]
local_ns: dict = {}
register_fixtures(
registry,
local_ns,
session_fixture="db_session",
strategy=LoadStrategy.SKIP_EXISTING,
)
inner = local_ns["fixture_new_roles"].__wrapped__ # type: ignore[attr-defined]
result = await inner(db_session=db_session)
assert len(result) == 1
assert result[0].id == role_id
@pytest.mark.anyio
async def test_skip_existing_returns_existing_record(
self, db_session: AsyncSession
):
"""SKIP_EXISTING returns the existing DB record when PK already present."""
role_id = uuid.uuid4()
existing = Role(id=role_id, name="already_there")
db_session.add(existing)
await db_session.flush()
registry = FixtureRegistry()
@registry.register()
def dup_roles() -> list[Role]:
return [Role(id=role_id, name="should_not_overwrite")]
local_ns: dict = {}
register_fixtures(
registry,
local_ns,
session_fixture="db_session",
strategy=LoadStrategy.SKIP_EXISTING,
)
inner = local_ns["fixture_dup_roles"].__wrapped__ # type: ignore[attr-defined]
result = await inner(db_session=db_session)
assert len(result) == 1
assert result[0].name == "already_there"
@pytest.mark.anyio
async def test_skip_existing_null_pk_inserts(self, db_session: AsyncSession):
"""SKIP_EXISTING with null PK (auto-increment) falls through to session.add()."""
registry = FixtureRegistry()
@registry.register()
def auto_roles() -> list[IntRole]:
return [IntRole(name="auto_int")]
local_ns: dict = {}
register_fixtures(
registry,
local_ns,
session_fixture="db_session",
strategy=LoadStrategy.SKIP_EXISTING,
)
inner = local_ns["fixture_auto_roles"].__wrapped__ # type: ignore[attr-defined]
result = await inner(db_session=db_session)
assert len(result) == 1
assert result[0].name == "auto_int"
class TestReloadWithRelationshipsCompositePK:
"""Integration test for _reload_with_relationships composite-PK fallback."""
@pytest.mark.anyio
async def test_composite_pk_fallback_loads_relationships(self):
"""Models with composite PKs are reloaded per-instance via session.get()."""
async with create_db_session(DATABASE_URL, _LocalBase) as session:
group = _Group(id=uuid.uuid4(), name="g1")
session.add(group)
await session.flush()
item = _CompositeItem(group_id=group.id, item_code="A")
session.add(item)
await session.flush()
load_opts = _relationship_load_options(_CompositeItem)
assert load_opts # _CompositeItem has 'group' relationship
reloaded = await _reload_with_relationships(session, [item], load_opts)
assert len(reloaded) == 1
reloaded_item = cast(_CompositeItem, reloaded[0])
assert reloaded_item.group is not None
assert reloaded_item.group.name == "g1"

110
uv.lock generated
View File

@@ -251,7 +251,7 @@ wheels = [
[[package]]
name = "fastapi-toolsets"
version = "3.0.0"
version = "3.1.0"
source = { editable = "." }
dependencies = [
{ name = "asyncpg" },
@@ -916,7 +916,7 @@ wheels = [
[[package]]
name = "pytest"
version = "9.0.2"
version = "9.0.3"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "colorama", marker = "sys_platform == 'win32'" },
@@ -925,9 +925,9 @@ dependencies = [
{ name = "pluggy" },
{ name = "pygments" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d1/db/7ef3487e0fb0049ddb5ce41d3a49c235bf9ad299b6a25d5780a89f19230f/pytest-9.0.2.tar.gz", hash = "sha256:75186651a92bd89611d1d9fc20f0b4345fd827c41ccd5c299a868a05d70edf11", size = 1568901, upload-time = "2025-12-06T21:30:51.014Z" }
sdist = { url = "https://files.pythonhosted.org/packages/7d/0d/549bd94f1a0a402dc8cf64563a117c0f3765662e2e668477624baeec44d5/pytest-9.0.3.tar.gz", hash = "sha256:b86ada508af81d19edeb213c681b1d48246c1a91d304c6c81a427674c17eb91c", size = 1572165, upload-time = "2026-04-07T17:16:18.027Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/3b/ab/b3226f0bd7cdcf710fbede2b3548584366da3b19b5021e74f5bde2a8fa3f/pytest-9.0.2-py3-none-any.whl", hash = "sha256:711ffd45bf766d5264d487b917733b453d917afd2b0ad65223959f59089f875b", size = 374801, upload-time = "2025-12-06T21:30:49.154Z" },
{ url = "https://files.pythonhosted.org/packages/d4/24/a372aaf5c9b7208e7112038812994107bc65a84cd00e0354a88c2c77a617/pytest-9.0.3-py3-none-any.whl", hash = "sha256:2c5efc453d45394fdd706ade797c0a81091eccd1d6e4bccfcd476e2b8e0ab5d9", size = 375249, upload-time = "2026-04-07T17:16:16.13Z" },
]
[[package]]
@@ -1064,27 +1064,27 @@ wheels = [
[[package]]
name = "ruff"
version = "0.15.8"
version = "0.15.9"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/14/b0/73cf7550861e2b4824950b8b52eebdcc5adc792a00c514406556c5b80817/ruff-0.15.8.tar.gz", hash = "sha256:995f11f63597ee362130d1d5a327a87cb6f3f5eae3094c620bcc632329a4d26e", size = 4610921, upload-time = "2026-03-26T18:39:38.675Z" }
sdist = { url = "https://files.pythonhosted.org/packages/e6/97/e9f1ca355108ef7194e38c812ef40ba98c7208f47b13ad78d023caa583da/ruff-0.15.9.tar.gz", hash = "sha256:29cbb1255a9797903f6dde5ba0188c707907ff44a9006eb273b5a17bfa0739a2", size = 4617361, upload-time = "2026-04-02T18:17:20.829Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/4a/92/c445b0cd6da6e7ae51e954939cb69f97e008dbe750cfca89b8cedc081be7/ruff-0.15.8-py3-none-linux_armv6l.whl", hash = "sha256:cbe05adeba76d58162762d6b239c9056f1a15a55bd4b346cfd21e26cd6ad7bc7", size = 10527394, upload-time = "2026-03-26T18:39:41.566Z" },
{ url = "https://files.pythonhosted.org/packages/eb/92/f1c662784d149ad1414cae450b082cf736430c12ca78367f20f5ed569d65/ruff-0.15.8-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:d3e3d0b6ba8dca1b7ef9ab80a28e840a20070c4b62e56d675c24f366ef330570", size = 10905693, upload-time = "2026-03-26T18:39:30.364Z" },
{ url = "https://files.pythonhosted.org/packages/ca/f2/7a631a8af6d88bcef997eb1bf87cc3da158294c57044aafd3e17030613de/ruff-0.15.8-py3-none-macosx_11_0_arm64.whl", hash = "sha256:6ee3ae5c65a42f273f126686353f2e08ff29927b7b7e203b711514370d500de3", size = 10323044, upload-time = "2026-03-26T18:39:33.37Z" },
{ url = "https://files.pythonhosted.org/packages/67/18/1bf38e20914a05e72ef3b9569b1d5c70a7ef26cd188d69e9ca8ef588d5bf/ruff-0.15.8-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:fdce027ada77baa448077ccc6ebb2fa9c3c62fd110d8659d601cf2f475858d94", size = 10629135, upload-time = "2026-03-26T18:39:44.142Z" },
{ url = "https://files.pythonhosted.org/packages/d2/e9/138c150ff9af60556121623d41aba18b7b57d95ac032e177b6a53789d279/ruff-0.15.8-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:12e617fc01a95e5821648a6df341d80456bd627bfab8a829f7cfc26a14a4b4a3", size = 10348041, upload-time = "2026-03-26T18:39:52.178Z" },
{ url = "https://files.pythonhosted.org/packages/02/f1/5bfb9298d9c323f842c5ddeb85f1f10ef51516ac7a34ba446c9347d898df/ruff-0.15.8-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:432701303b26416d22ba696c39f2c6f12499b89093b61360abc34bcc9bf07762", size = 11121987, upload-time = "2026-03-26T18:39:55.195Z" },
{ url = "https://files.pythonhosted.org/packages/10/11/6da2e538704e753c04e8d86b1fc55712fdbdcc266af1a1ece7a51fff0d10/ruff-0.15.8-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:d910ae974b7a06a33a057cb87d2a10792a3b2b3b35e33d2699fdf63ec8f6b17a", size = 11951057, upload-time = "2026-03-26T18:39:19.18Z" },
{ url = "https://files.pythonhosted.org/packages/83/f0/c9208c5fd5101bf87002fed774ff25a96eea313d305f1e5d5744698dc314/ruff-0.15.8-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:2033f963c43949d51e6fdccd3946633c6b37c484f5f98c3035f49c27395a8ab8", size = 11464613, upload-time = "2026-03-26T18:40:06.301Z" },
{ url = "https://files.pythonhosted.org/packages/f8/22/d7f2fabdba4fae9f3b570e5605d5eb4500dcb7b770d3217dca4428484b17/ruff-0.15.8-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:0f29b989a55572fb885b77464cf24af05500806ab4edf9a0fd8977f9759d85b1", size = 11257557, upload-time = "2026-03-26T18:39:57.972Z" },
{ url = "https://files.pythonhosted.org/packages/71/8c/382a9620038cf6906446b23ce8632ab8c0811b8f9d3e764f58bedd0c9a6f/ruff-0.15.8-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:ac51d486bf457cdc985a412fb1801b2dfd1bd8838372fc55de64b1510eff4bec", size = 11169440, upload-time = "2026-03-26T18:39:22.205Z" },
{ url = "https://files.pythonhosted.org/packages/4d/0d/0994c802a7eaaf99380085e4e40c845f8e32a562e20a38ec06174b52ef24/ruff-0.15.8-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:c9861eb959edab053c10ad62c278835ee69ca527b6dcd72b47d5c1e5648964f6", size = 10605963, upload-time = "2026-03-26T18:39:46.682Z" },
{ url = "https://files.pythonhosted.org/packages/19/aa/d624b86f5b0aad7cef6bbf9cd47a6a02dfdc4f72c92a337d724e39c9d14b/ruff-0.15.8-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:8d9a5b8ea13f26ae90838afc33f91b547e61b794865374f114f349e9036835fb", size = 10357484, upload-time = "2026-03-26T18:39:49.176Z" },
{ url = "https://files.pythonhosted.org/packages/35/c3/e0b7835d23001f7d999f3895c6b569927c4d39912286897f625736e1fd04/ruff-0.15.8-py3-none-musllinux_1_2_i686.whl", hash = "sha256:c2a33a529fb3cbc23a7124b5c6ff121e4d6228029cba374777bd7649cc8598b8", size = 10830426, upload-time = "2026-03-26T18:40:03.702Z" },
{ url = "https://files.pythonhosted.org/packages/f0/51/ab20b322f637b369383adc341d761eaaa0f0203d6b9a7421cd6e783d81b9/ruff-0.15.8-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:75e5cd06b1cf3f47a3996cfc999226b19aa92e7cce682dcd62f80d7035f98f49", size = 11345125, upload-time = "2026-03-26T18:39:27.799Z" },
{ url = "https://files.pythonhosted.org/packages/37/e6/90b2b33419f59d0f2c4c8a48a4b74b460709a557e8e0064cf33ad894f983/ruff-0.15.8-py3-none-win32.whl", hash = "sha256:bc1f0a51254ba21767bfa9a8b5013ca8149dcf38092e6a9eb704d876de94dc34", size = 10571959, upload-time = "2026-03-26T18:39:36.117Z" },
{ url = "https://files.pythonhosted.org/packages/1f/a2/ef467cb77099062317154c63f234b8a7baf7cb690b99af760c5b68b9ee7f/ruff-0.15.8-py3-none-win_amd64.whl", hash = "sha256:04f79eff02a72db209d47d665ba7ebcad609d8918a134f86cb13dd132159fc89", size = 11743893, upload-time = "2026-03-26T18:39:25.01Z" },
{ url = "https://files.pythonhosted.org/packages/15/e2/77be4fff062fa78d9b2a4dea85d14785dac5f1d0c1fb58ed52331f0ebe28/ruff-0.15.8-py3-none-win_arm64.whl", hash = "sha256:cf891fa8e3bb430c0e7fac93851a5978fc99c8fa2c053b57b118972866f8e5f2", size = 11048175, upload-time = "2026-03-26T18:40:01.06Z" },
{ url = "https://files.pythonhosted.org/packages/0b/1f/9cdfd0ac4b9d1e5a6cf09bedabdf0b56306ab5e333c85c87281273e7b041/ruff-0.15.9-py3-none-linux_armv6l.whl", hash = "sha256:6efbe303983441c51975c243e26dff328aca11f94b70992f35b093c2e71801e1", size = 10511206, upload-time = "2026-04-02T18:16:41.574Z" },
{ url = "https://files.pythonhosted.org/packages/3d/f6/32bfe3e9c136b35f02e489778d94384118bb80fd92c6d92e7ccd97db12ce/ruff-0.15.9-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:4965bac6ac9ea86772f4e23587746f0b7a395eccabb823eb8bfacc3fa06069f7", size = 10923307, upload-time = "2026-04-02T18:17:08.645Z" },
{ url = "https://files.pythonhosted.org/packages/ca/25/de55f52ab5535d12e7aaba1de37a84be6179fb20bddcbe71ec091b4a3243/ruff-0.15.9-py3-none-macosx_11_0_arm64.whl", hash = "sha256:eaf05aad70ca5b5a0a4b0e080df3a6b699803916d88f006efd1f5b46302daab8", size = 10316722, upload-time = "2026-04-02T18:16:44.206Z" },
{ url = "https://files.pythonhosted.org/packages/48/11/690d75f3fd6278fe55fff7c9eb429c92d207e14b25d1cae4064a32677029/ruff-0.15.9-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:9439a342adb8725f32f92732e2bafb6d5246bd7a5021101166b223d312e8fc59", size = 10623674, upload-time = "2026-04-02T18:16:50.951Z" },
{ url = "https://files.pythonhosted.org/packages/bd/ec/176f6987be248fc5404199255522f57af1b4a5a1b57727e942479fec98ad/ruff-0.15.9-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:9c5e6faf9d97c8edc43877c3f406f47446fc48c40e1442d58cfcdaba2acea745", size = 10351516, upload-time = "2026-04-02T18:16:57.206Z" },
{ url = "https://files.pythonhosted.org/packages/b2/fc/51cffbd2b3f240accc380171d51446a32aa2ea43a40d4a45ada67368fbd2/ruff-0.15.9-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:7b34a9766aeec27a222373d0b055722900fbc0582b24f39661aa96f3fe6ad901", size = 11150202, upload-time = "2026-04-02T18:17:06.452Z" },
{ url = "https://files.pythonhosted.org/packages/d6/d4/25292a6dfc125f6b6528fe6af31f5e996e19bf73ca8e3ce6eb7fa5b95885/ruff-0.15.9-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:89dd695bc72ae76ff484ae54b7e8b0f6b50f49046e198355e44ea656e521fef9", size = 11988891, upload-time = "2026-04-02T18:17:18.575Z" },
{ url = "https://files.pythonhosted.org/packages/13/e1/1eebcb885c10e19f969dcb93d8413dfee8172578709d7ee933640f5e7147/ruff-0.15.9-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:ce187224ef1de1bd225bc9a152ac7102a6171107f026e81f317e4257052916d5", size = 11480576, upload-time = "2026-04-02T18:16:52.986Z" },
{ url = "https://files.pythonhosted.org/packages/ff/6b/a1548ac378a78332a4c3dcf4a134c2475a36d2a22ddfa272acd574140b50/ruff-0.15.9-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2b0c7c341f68adb01c488c3b7d4b49aa8ea97409eae6462d860a79cf55f431b6", size = 11254525, upload-time = "2026-04-02T18:17:02.041Z" },
{ url = "https://files.pythonhosted.org/packages/42/aa/4bb3af8e61acd9b1281db2ab77e8b2c3c5e5599bf2a29d4a942f1c62b8d6/ruff-0.15.9-py3-none-manylinux_2_31_riscv64.whl", hash = "sha256:55cc15eee27dc0eebdfcb0d185a6153420efbedc15eb1d38fe5e685657b0f840", size = 11204072, upload-time = "2026-04-02T18:17:13.581Z" },
{ url = "https://files.pythonhosted.org/packages/69/48/d550dc2aa6e423ea0bcc1d0ff0699325ffe8a811e2dba156bd80750b86dc/ruff-0.15.9-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:a6537f6eed5cda688c81073d46ffdfb962a5f29ecb6f7e770b2dc920598997ed", size = 10594998, upload-time = "2026-04-02T18:16:46.369Z" },
{ url = "https://files.pythonhosted.org/packages/63/47/321167e17f5344ed5ec6b0aa2cff64efef5f9e985af8f5622cfa6536043f/ruff-0.15.9-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:6d3fcbca7388b066139c523bda744c822258ebdcfbba7d24410c3f454cc9af71", size = 10359769, upload-time = "2026-04-02T18:17:10.994Z" },
{ url = "https://files.pythonhosted.org/packages/67/5e/074f00b9785d1d2c6f8c22a21e023d0c2c1817838cfca4c8243200a1fa87/ruff-0.15.9-py3-none-musllinux_1_2_i686.whl", hash = "sha256:058d8e99e1bfe79d8a0def0b481c56059ee6716214f7e425d8e737e412d69677", size = 10850236, upload-time = "2026-04-02T18:16:48.749Z" },
{ url = "https://files.pythonhosted.org/packages/76/37/804c4135a2a2caf042925d30d5f68181bdbd4461fd0d7739da28305df593/ruff-0.15.9-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:8e1ddb11dbd61d5983fa2d7d6370ef3eb210951e443cace19594c01c72abab4c", size = 11358343, upload-time = "2026-04-02T18:16:55.068Z" },
{ url = "https://files.pythonhosted.org/packages/88/3d/1364fcde8656962782aa9ea93c92d98682b1ecec2f184e625a965ad3b4a6/ruff-0.15.9-py3-none-win32.whl", hash = "sha256:bde6ff36eaf72b700f32b7196088970bf8fdb2b917b7accd8c371bfc0fd573ec", size = 10583382, upload-time = "2026-04-02T18:17:04.261Z" },
{ url = "https://files.pythonhosted.org/packages/4c/56/5c7084299bd2cacaa07ae63a91c6f4ba66edc08bf28f356b24f6b717c799/ruff-0.15.9-py3-none-win_amd64.whl", hash = "sha256:45a70921b80e1c10cf0b734ef09421f71b5aa11d27404edc89d7e8a69505e43d", size = 11744969, upload-time = "2026-04-02T18:16:59.611Z" },
{ url = "https://files.pythonhosted.org/packages/03/36/76704c4f312257d6dbaae3c959add2a622f63fcca9d864659ce6d8d97d3d/ruff-0.15.9-py3-none-win_arm64.whl", hash = "sha256:0694e601c028fd97dc5c6ee244675bc241aeefced7ef80cd9c6935a871078f53", size = 11005870, upload-time = "2026-04-02T18:17:15.773Z" },
]
[[package]]
@@ -1228,26 +1228,26 @@ wheels = [
[[package]]
name = "ty"
version = "0.0.27"
version = "0.0.29"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/f4/de/e5cf1f151cf52fe1189e42d03d90909d7d1354fdc0c1847cbb63a0baa3da/ty-0.0.27.tar.gz", hash = "sha256:d7a8de3421d92420b40c94fe7e7d4816037560621903964dd035cf9bd0204a73", size = 5424130, upload-time = "2026-03-31T19:07:20.806Z" }
sdist = { url = "https://files.pythonhosted.org/packages/47/d5/853561de49fae38c519e905b2d8da9c531219608f1fccc47a0fc2c896980/ty-0.0.29.tar.gz", hash = "sha256:e7936cca2f691eeda631876c92809688dbbab68687c3473f526cd83b6a9228d8", size = 5469221, upload-time = "2026-04-05T15:01:21.328Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/fa/20/2a9ea661758bd67f2bfd54ce9daacb5a26c56c5f8b49fbd9a43b365a8a7d/ty-0.0.27-py3-none-linux_armv6l.whl", hash = "sha256:eb14456b8611c9e8287aa9b633f4d2a0d9f3082a31796969e0b50bdda8930281", size = 10571211, upload-time = "2026-03-31T19:07:23.28Z" },
{ url = "https://files.pythonhosted.org/packages/da/b2/8887a51f705d075ddbe78ae7f0d4755ef48d0a90235f67aee289e9cee950/ty-0.0.27-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:02e662184703db7586118df611cf24a000d35dae38d950053d1dd7b6736fd2c4", size = 10427576, upload-time = "2026-03-31T19:07:15.499Z" },
{ url = "https://files.pythonhosted.org/packages/1d/c3/79d88163f508fb709ce19bc0b0a66c7c64b53d372d4caa56172c3d9b3ae8/ty-0.0.27-py3-none-macosx_11_0_arm64.whl", hash = "sha256:be5fc2899441f7f8f7ef40f9ffd006075a5ff6b06c44e8d2aa30e1b900c12f51", size = 9870359, upload-time = "2026-03-31T19:07:36.852Z" },
{ url = "https://files.pythonhosted.org/packages/dc/4d/ed1b0db0e1e46b5ed4976bbfe0d1825faf003b4e3774ef28c785ed73e4bb/ty-0.0.27-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:30231e652b14742a76b64755e54bf0cb1cd4c128bcaf625222e0ca92a2094887", size = 10380488, upload-time = "2026-03-31T19:07:31.268Z" },
{ url = "https://files.pythonhosted.org/packages/b1/f2/20372f6d510b01570028433064880adec2f8abe68bf0c4603be61a560bef/ty-0.0.27-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:5a119b1168f64261b3205a37e40b5b6c4aac8fd58e4587988f4e4b22c3c79847", size = 10390248, upload-time = "2026-03-31T19:07:28.345Z" },
{ url = "https://files.pythonhosted.org/packages/45/4b/46b31a7311306be1a560f7f20fdc37b5bf718787f60626cd265d9b637554/ty-0.0.27-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:e38f4e187b6975d2cbebf0f1eb1221f8f64f6e509bad14d7bb2a91afc97e4956", size = 10878479, upload-time = "2026-03-31T19:07:39.393Z" },
{ url = "https://files.pythonhosted.org/packages/42/ba/5231a2a1fb1cebe053a25de8fded95e1a30a1e77d3628a9e58487297bafc/ty-0.0.27-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:a07b1a8fbb23844f6d22091275430d9ac617175f34aa99159b268193de210389", size = 11461232, upload-time = "2026-03-31T19:07:02.518Z" },
{ url = "https://files.pythonhosted.org/packages/c3/37/558abab3e1f6670493524f61280b4dfcc3219555f13889223e733381dfab/ty-0.0.27-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:d3ec4033031f240836bb0337274bac5c49dde312c7c6d7575451ed719bf8ffa3", size = 11133002, upload-time = "2026-03-31T19:07:18.371Z" },
{ url = "https://files.pythonhosted.org/packages/32/38/188c14a57f52160407ce62c6abb556011718fd0bcbe1dca690529ce84c46/ty-0.0.27-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:924a8849afd500d260bf5b7296165a05b7424fbb6b19113f30f3b999d682873f", size = 10986624, upload-time = "2026-03-31T19:07:13.066Z" },
{ url = "https://files.pythonhosted.org/packages/9f/f1/667a71393f47d2cd6ba9ed07541b8df3eb63aab1f2ee658e77d91b8362fa/ty-0.0.27-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:d8270026c07e7423a1b3a3fd065b46ed1478748f0662518b523b57744f3fa025", size = 10366721, upload-time = "2026-03-31T19:07:00.131Z" },
{ url = "https://files.pythonhosted.org/packages/8b/aa/8edafe41be898bda774249abc5be6edd733e53fb1777d59ea9331e38537d/ty-0.0.27-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:e26e9735d3bdfd95d881111ad1cf570eab8188d8c3be36d6bcaad044d38984d8", size = 10412239, upload-time = "2026-03-31T19:07:05.297Z" },
{ url = "https://files.pythonhosted.org/packages/53/ff/8bafaed4a18d38264f46bdfc427de7ea2974cf9064e4e0bdb1b6e6c724e3/ty-0.0.27-py3-none-musllinux_1_2_i686.whl", hash = "sha256:7c09cc9a699810609acc0090af8d0db68adaee6e60a7c3e05ab80cc954a83db7", size = 10573507, upload-time = "2026-03-31T19:06:57.064Z" },
{ url = "https://files.pythonhosted.org/packages/16/2e/63a8284a2fefd08ab56ecbad0fde7dd4b2d4045a31cf24c1d1fcd9643227/ty-0.0.27-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:2d3e02853bb037221a456e034b1898aaa573e6374fbb53884e33cb7513ccb85a", size = 11090233, upload-time = "2026-03-31T19:07:34.139Z" },
{ url = "https://files.pythonhosted.org/packages/14/d3/d6fa1cafdfa2b34dbfa304fc6833af8e1669fc34e24d214fa76d2a2e5a25/ty-0.0.27-py3-none-win32.whl", hash = "sha256:34e7377f2047c14dbbb7bf5322e84114db7a5f2cb470db6bee63f8f3550cfc1e", size = 9984415, upload-time = "2026-03-31T19:07:07.98Z" },
{ url = "https://files.pythonhosted.org/packages/85/e6/dd4e27da9632b3472d5711ca49dbd3709dbd3e8c73f3af6db9c254235ca9/ty-0.0.27-py3-none-win_amd64.whl", hash = "sha256:3f7e4145aad8b815ed69b324c93b5b773eb864dda366ca16ab8693ff88ce6f36", size = 10961535, upload-time = "2026-03-31T19:07:10.566Z" },
{ url = "https://files.pythonhosted.org/packages/0e/1a/824b3496d66852ed7d5d68d9787711131552b68dce8835ce9410db32e618/ty-0.0.27-py3-none-win_arm64.whl", hash = "sha256:95bf8d01eb96bb2ba3ffc39faff19da595176448e80871a7b362f4d2de58476c", size = 10376689, upload-time = "2026-03-31T19:07:25.732Z" },
{ url = "https://files.pythonhosted.org/packages/03/b7/911f9962115acfa24e3b2ec9d4992dd994c38e8769e1b1d7680bb4d28a51/ty-0.0.29-py3-none-linux_armv6l.whl", hash = "sha256:b8a40955f7660d3eaceb0d964affc81b790c0765e7052921a5f861ff8a471c30", size = 10568206, upload-time = "2026-04-05T15:01:19.165Z" },
{ url = "https://files.pythonhosted.org/packages/fe/c3/fcae2167d4c77a97269f92f11d1b43b03617f81de1283d5d05b43432110c/ty-0.0.29-py3-none-macosx_10_12_x86_64.whl", hash = "sha256:6b6849adae15b00bbe2d3c5b078967dcb62eba37d38936b8eeb4c81a82d2e3b8", size = 10442530, upload-time = "2026-04-05T15:01:28.471Z" },
{ url = "https://files.pythonhosted.org/packages/97/33/5a6bfa240cfcb9c36046ae2459fa9ea23238d20130d8656ff5ac4d6c012a/ty-0.0.29-py3-none-macosx_11_0_arm64.whl", hash = "sha256:dcdd9b17209788152f7b7ea815eda07989152325052fe690013537cc7904ce49", size = 9915735, upload-time = "2026-04-05T15:01:10.365Z" },
{ url = "https://files.pythonhosted.org/packages/b3/1e/318f45fae232118e81a6306c30f50de42c509c412128d5bd231eab699ffb/ty-0.0.29-py3-none-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:9d8ed4789bae78ffaf94462c0d25589a734cab0366b86f2bbcb1bb90e1a7a169", size = 10419748, upload-time = "2026-04-05T15:01:32.375Z" },
{ url = "https://files.pythonhosted.org/packages/a9/a8/5687872e2ab5a0f7dd4fd8456eac31e9381ad4dc74961f6f29965ad4dd91/ty-0.0.29-py3-none-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:91ec374b8565e0ad0900011c24641ebbef2da51adbd4fb69ff3280c8a7eceb02", size = 10394738, upload-time = "2026-04-05T15:01:06.473Z" },
{ url = "https://files.pythonhosted.org/packages/de/68/015d118097eeb95e6a44c4abce4c0a28b7b9dfb3085b7f0ee48e4f099633/ty-0.0.29-py3-none-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:298a8d5faa2502d3810bbbb47a030b9455495b9921594206043c785dd61548cf", size = 10910613, upload-time = "2026-04-05T15:01:17.17Z" },
{ url = "https://files.pythonhosted.org/packages/1c/01/47ce3c6c53e0670eadbe80756b167bf80ed6681d1ba57cfde2e8065a13d1/ty-0.0.29-py3-none-manylinux_2_17_ppc64le.manylinux2014_ppc64le.whl", hash = "sha256:3c8fba1a3524c6109d1e020d92301c79d41bf442fa8d335b9fa366239339cb70", size = 11475750, upload-time = "2026-04-05T15:01:30.461Z" },
{ url = "https://files.pythonhosted.org/packages/c4/cf/e361845b1081c9264ad5b7c963231bab03f2666865a9f2a115c4233f2137/ty-0.0.29-py3-none-manylinux_2_17_s390x.manylinux2014_s390x.whl", hash = "sha256:4c48adf88a70d264128c39ee922ed14a947817fced1e93c08c1a89c9244edcde", size = 11190055, upload-time = "2026-04-05T15:01:12.369Z" },
{ url = "https://files.pythonhosted.org/packages/79/12/0fb0857e9a62cb11586e9a712103877bbf717f5fb570d16634408cfdefee/ty-0.0.29-py3-none-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:2ce0a7a0e96bc7b42518cd3a1a6a6298ef64ff40ca4614355c1aa807059b5c6f", size = 11020539, upload-time = "2026-04-05T15:01:37.022Z" },
{ url = "https://files.pythonhosted.org/packages/20/36/5a26753802083f80cd125db6c4348ad42b3c982ec36e718e0bf4c18f75e5/ty-0.0.29-py3-none-musllinux_1_2_aarch64.whl", hash = "sha256:a6ac86a05b4a3731d45365ab97780acc7b8146fa62fccb3cbe94fe6546c67a97", size = 10396399, upload-time = "2026-04-05T15:01:26.167Z" },
{ url = "https://files.pythonhosted.org/packages/00/e6/b4e75b5752239ab3ab400f19faef4dbef81d05aab5d3419fda0c062a3765/ty-0.0.29-py3-none-musllinux_1_2_armv7l.whl", hash = "sha256:6bbbf53141af0f3150bf288d716263f1a3550054e4b3551ca866d38192ba9891", size = 10421461, upload-time = "2026-04-05T15:01:08.367Z" },
{ url = "https://files.pythonhosted.org/packages/c0/21/1084b5b609f9abed62070ec0b31c283a403832a6310c8bbc208bd45ee1e6/ty-0.0.29-py3-none-musllinux_1_2_i686.whl", hash = "sha256:1c9e06b770c1d0ff5efc51e34312390db31d53fcf3088163f413030b42b74f84", size = 10599187, upload-time = "2026-04-05T15:01:23.52Z" },
{ url = "https://files.pythonhosted.org/packages/ab/a1/ce19a2ca717bbcc1ee11378aba52ef70b6ce5b87245162a729d9fdc2360f/ty-0.0.29-py3-none-musllinux_1_2_x86_64.whl", hash = "sha256:0307fe37e3f000ef1a4ae230bbaf511508a78d24a5e51b40902a21b09d5e6037", size = 11121198, upload-time = "2026-04-05T15:01:15.22Z" },
{ url = "https://files.pythonhosted.org/packages/6b/6b/f1430b279af704321566ce7ec2725d3d8258c2f815ebd93e474c64cd4543/ty-0.0.29-py3-none-win32.whl", hash = "sha256:7a2a898217960a825f8bc0087e1fdbaf379606175e98f9807187221d53a4a8ed", size = 9995331, upload-time = "2026-04-05T15:01:01.32Z" },
{ url = "https://files.pythonhosted.org/packages/d2/ef/3ef01c17785ff9a69378465c7d0faccd48a07b163554db0995e5d65a5a23/ty-0.0.29-py3-none-win_amd64.whl", hash = "sha256:fc1294200226b91615acbf34e0a9ad81caf98c081e9c6a912a31b0a7b603bc3f", size = 11023644, upload-time = "2026-04-05T15:01:04.432Z" },
{ url = "https://files.pythonhosted.org/packages/2c/55/87280a994d6a2d2647c65e12abbc997ed49835794366153c04c4d9304d76/ty-0.0.29-py3-none-win_arm64.whl", hash = "sha256:f9794bbd1bb3ce13f78c191d0c89ae4c63f52c12b6daa0c6fe220b90d019d12c", size = 10428165, upload-time = "2026-04-05T15:01:34.665Z" },
]
[[package]]
@@ -1324,7 +1324,7 @@ wheels = [
[[package]]
name = "zensical"
version = "0.0.31"
version = "0.0.32"
source = { registry = "https://pypi.org/simple" }
dependencies = [
{ name = "click" },
@@ -1334,18 +1334,18 @@ dependencies = [
{ name = "pymdown-extensions" },
{ name = "pyyaml" },
]
sdist = { url = "https://files.pythonhosted.org/packages/d5/1a/9b6f5285c5aef648db38f9132f49a7059bd2c9d748f68ef0c52ed8afcff3/zensical-0.0.31.tar.gz", hash = "sha256:9c12f07bde70c4bfdb13d6cae1bedf8d18064d257a6e81128a152502b28a8fc3", size = 3891758, upload-time = "2026-04-01T11:30:21.88Z" }
sdist = { url = "https://files.pythonhosted.org/packages/7a/94/4a49ca9329136445f4111fda60e4bfcbe68d95e18e9aa02e4606fba5df4a/zensical-0.0.32.tar.gz", hash = "sha256:0f857b09a2b10c99202b3712e1ffc4d1d1ffa4c7c2f1aa0fafb1346b2d8df604", size = 3891955, upload-time = "2026-04-07T11:41:29.203Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/c2/db/cc4e555d2e816f2d91304ff969d62cc3a401ee477dbb7c720b874bec67d6/zensical-0.0.31-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:b489936d670733dd204f16b689a2acc0e45b69e42cc4901f5131ae57658b8fbc", size = 12419980, upload-time = "2026-04-01T11:29:44.01Z" },
{ url = "https://files.pythonhosted.org/packages/e7/c1/6789f73164c7f5821f5defb8a80b1dba8d5af24bdec7db36876793c5afd9/zensical-0.0.31-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:d9f678efc0d9918e45eeb8bc62847b2cce23db7393c8c59c1be6d1c064bbaacd", size = 12292301, upload-time = "2026-04-01T11:29:47.277Z" },
{ url = "https://files.pythonhosted.org/packages/4f/9a/6a83ad209081a953e0285d5056e5452c4fbcabd2f104f3797d53e4bdd96f/zensical-0.0.31-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:fb2b50ecf674997f818e53f12f2a67875a21b0c79ed74c151dfaef2f1475e5bf", size = 12661472, upload-time = "2026-04-01T11:29:50.706Z" },
{ url = "https://files.pythonhosted.org/packages/9c/4a/a82f5c81893b7a607cf9d439b75c3c3894b4ef4d3e92d5d818b4fa5c6f23/zensical-0.0.31-cp310-abi3-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:6fb5c634fe88254770a2d4db5c05b06f1c3ee5e29d2ae3e7efdae8905e435b1d", size = 12603784, upload-time = "2026-04-01T11:29:53.623Z" },
{ url = "https://files.pythonhosted.org/packages/f7/1c/79c198628b8e006be32dfb1c5b73561757a349a6cf3069600a67ffa62495/zensical-0.0.31-cp310-abi3-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:94e64630552793274db1ec66c971e49a15ad351536d5d12de67ec6da7358ac50", size = 12959832, upload-time = "2026-04-01T11:29:56.736Z" },
{ url = "https://files.pythonhosted.org/packages/db/9d/45839d9ca0f69622e8a3b944f2d8d7f7d2b7c2da78201079c4feb275feb6/zensical-0.0.31-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:738a2fd5832e3b3c10ff642eebaf89c89ca1d28e4451dad0f36fdac53c415577", size = 12704024, upload-time = "2026-04-01T11:29:59.836Z" },
{ url = "https://files.pythonhosted.org/packages/df/5f/451d7f4d94092bc38bd8d514826fb7b0329c188db506795b1d20bd07d517/zensical-0.0.31-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:bd601f6132e285ef6c3e4c3852be2094fc0473295a8080003db76a79760f84fb", size = 12837788, upload-time = "2026-04-01T11:30:03.048Z" },
{ url = "https://files.pythonhosted.org/packages/d8/39/390a8fc384fb174ebd4450343a0aa2877b3a31ddcedf5ef0b8d26944e12c/zensical-0.0.31-cp310-abi3-musllinux_1_2_armv7l.whl", hash = "sha256:dc3b6a9dfb5903c0aa779ef65cd6185add2b8aa1db237be840874b8c9db761b8", size = 12876822, upload-time = "2026-04-01T11:30:06.418Z" },
{ url = "https://files.pythonhosted.org/packages/d5/60/640da2f095782cf38974cd851fb7afa62651d09a36543a1d8942b31aabdc/zensical-0.0.31-cp310-abi3-musllinux_1_2_i686.whl", hash = "sha256:ddd4321b275e82c4897aa45b05038ce204b88fb311ad55f8c2af572173a9b56c", size = 13024036, upload-time = "2026-04-01T11:30:09.501Z" },
{ url = "https://files.pythonhosted.org/packages/3f/06/0564377cbfccea3653254adfa851c1b20d1696e4b16770c7b2e1dd1ef1d7/zensical-0.0.31-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:147ab4bc17f3088f703aa6c4b9c416411f4ea8ca64d26f6586beae49d97fd3c7", size = 12975505, upload-time = "2026-04-01T11:30:12.268Z" },
{ url = "https://files.pythonhosted.org/packages/35/4b/b8a0c4e5937cb05882dcce667798403e00897135080a69f92363e5e3ff9f/zensical-0.0.31-cp310-abi3-win32.whl", hash = "sha256:03fa11e629a308507693489541f43e751697784e94365e7435b02104aefd1c2c", size = 12011233, upload-time = "2026-04-01T11:30:15.496Z" },
{ url = "https://files.pythonhosted.org/packages/3e/99/0eacdb466d344c0c86596932201268517be42f3e0bb6c78b2b0cd84c55f6/zensical-0.0.31-cp310-abi3-win_amd64.whl", hash = "sha256:d6621d4bb46af4143560045d4a18c8c76302db56bf1dbb6e2ce107d7fb643e09", size = 12207545, upload-time = "2026-04-01T11:30:19.054Z" },
{ url = "https://files.pythonhosted.org/packages/73/e1/dd03762447f1c2a4c8aff08e8f047ec17c73421714a0600ef71c361a5934/zensical-0.0.32-cp310-abi3-macosx_10_12_x86_64.whl", hash = "sha256:7ed181c76c03fec4c2dd5db207810044bf9c3fa87097fbdbabd633661e20fc70", size = 12416474, upload-time = "2026-04-07T11:40:55.888Z" },
{ url = "https://files.pythonhosted.org/packages/f5/a6/2f1babb00842c6efa5ae755b3ab414e4688ae8e47bdd2e785c0c37ef625d/zensical-0.0.32-cp310-abi3-macosx_11_0_arm64.whl", hash = "sha256:8cde82bf256408f75ae2b07bffcaac7d080b6aad5f7acf210c438cb7413c3081", size = 12292801, upload-time = "2026-04-07T11:40:59.648Z" },
{ url = "https://files.pythonhosted.org/packages/2d/f1/d32706de06fd30fb07ae514222a79dd17d4578cd1634e5b692e0c790a61e/zensical-0.0.32-cp310-abi3-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:60e60e2358249b2a2c5e1c5c04586d8dbba27e577441cc9dd32fe8d879c6951e", size = 12658847, upload-time = "2026-04-07T11:41:02.347Z" },
{ url = "https://files.pythonhosted.org/packages/e7/42/a3daf4047c86382749a59795c4e7acd59952b4f6f37f329cd2d41cc37a0f/zensical-0.0.32-cp310-abi3-manylinux_2_17_armv7l.manylinux2014_armv7l.whl", hash = "sha256:ec79b4304009138e7a38ebe24e8a8e9dbc15d38922185f8a84470a7757d7b73f", size = 12604777, upload-time = "2026-04-07T11:41:05.227Z" },
{ url = "https://files.pythonhosted.org/packages/59/11/4af61d3fb07713cd3f77981c1b3017a60c2b210b36f1b04353f9116d03ca/zensical-0.0.32-cp310-abi3-manylinux_2_17_i686.manylinux2014_i686.whl", hash = "sha256:fc92fa7d0860ec6d95426a5f545cfc5493c60f8ab44fcc11611a4251f34f1b70", size = 12956242, upload-time = "2026-04-07T11:41:07.58Z" },
{ url = "https://files.pythonhosted.org/packages/8c/34/e9b5f4376bbf460f8c07a77af59bd169c7c68ed719a074e6667ba41109f8/zensical-0.0.32-cp310-abi3-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:07f69019396060e310c9c3b18747ce8982ad56d67fbab269b61e74a6a5bdcb4a", size = 12701954, upload-time = "2026-04-07T11:41:10.532Z" },
{ url = "https://files.pythonhosted.org/packages/d2/43/a52e5dcb324f38a1d22f7fafd4eec273385d04de52a7ab5ac7b444cf2bdc/zensical-0.0.32-cp310-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:d096c9ed20a48e5ff095eca218eef94f67e739cdf0abf7e1f7e232e78f6d980c", size = 12835464, upload-time = "2026-04-07T11:41:13.152Z" },
{ url = "https://files.pythonhosted.org/packages/a7/95/bede89ecb4932bbd29db7b61bf530a962aed09d3a8d5aa71a64af1d4920f/zensical-0.0.32-cp310-abi3-musllinux_1_2_armv7l.whl", hash = "sha256:bf5576b7154bde18cebd9a7b065d3ab8b334c6e73d5b2e83abe2b17f9d00a992", size = 12876574, upload-time = "2026-04-07T11:41:16.085Z" },
{ url = "https://files.pythonhosted.org/packages/9e/e8/9b25fda22bf729ca2598cc42cefe9b20e751d12d23e35c70ea0c7939d20a/zensical-0.0.32-cp310-abi3-musllinux_1_2_i686.whl", hash = "sha256:f33905a1e0b03a2ad548554a157b7f7c398e6f41012d1e755105ae2bc60eab8a", size = 13022702, upload-time = "2026-04-07T11:41:18.947Z" },
{ url = "https://files.pythonhosted.org/packages/f6/35/0c6d0b57187bd470a05e8a391c0edd1d690eb429e12b9755c99cf60a370e/zensical-0.0.32-cp310-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:0a73a53b1dd41fd239875a3cb57c4284747989c45b6933f18e9b51f1b5f3d8ef", size = 12975593, upload-time = "2026-04-07T11:41:21.436Z" },
{ url = "https://files.pythonhosted.org/packages/ee/2d/4e88bcefc33b7af22f0637fd002d3cf5384e8354f0a7f8a9dbfcd40cfa24/zensical-0.0.32-cp310-abi3-win32.whl", hash = "sha256:f8cb579bdb9b56f1704b93f4e17b42895c8cb466e8eec933fbe0153b5b1e3459", size = 12012163, upload-time = "2026-04-07T11:41:23.975Z" },
{ url = "https://files.pythonhosted.org/packages/8a/ae/a80a2f15fd10201fe3dfd6b5cdf85351165f820cf5b29e3c3b24092c158c/zensical-0.0.32-cp310-abi3-win_amd64.whl", hash = "sha256:6d662f42b5d0eadfac6d281e9d86574bc7a9f812f1ed496335d15f2d581d4b28", size = 12205948, upload-time = "2026-04-07T11:41:27.056Z" },
]

View File

@@ -2,10 +2,15 @@
site_name = "FastAPI Toolsets"
site_description = "Production-ready utilities for FastAPI applications."
site_author = "d3vyce"
site_url = "https://fastapi-toolsets.d3vyce.fr"
site_url = "https://fastapi-toolsets.d3vyce.fr/"
copyright = "Copyright &copy; 2026 d3vyce"
repo_url = "https://github.com/d3vyce/fastapi-toolsets"
[project.extra.version]
provider = "mike"
default = "stable"
alias = true
[project.theme]
custom_dir = "docs/overrides"
language = "en"