From 81c2a21891965e5f92c537e7f413aa4df78ed5d1 Mon Sep 17 00:00:00 2001 From: Claude Date: Tue, 27 Oct 2026 16:27:15 +0000 Subject: [PATCH] feat: add bulk delete, update and custom actions to AsyncCrud --- README.md | 2 +- docs/module/crud.md | 82 ++++ docs/module/exceptions.md | 1 + docs/reference/exceptions.md | 3 + docs/reference/schemas.md | 12 + src/fastapi_toolsets/crud/factory.py | 382 ++++++++++++++++- src/fastapi_toolsets/exceptions/__init__.py | 2 + src/fastapi_toolsets/exceptions/exceptions.py | 24 ++ src/fastapi_toolsets/schemas.py | 57 ++- src/fastapi_toolsets/types.py | 1 + tests/test_crud_bulk.py | 385 ++++++++++++++++++ 11 files changed, 937 insertions(+), 14 deletions(-) create mode 100644 tests/test_crud_bulk.py diff --git a/README.md b/README.md index 838acb05..5556c4d2 100644 --- a/README.md +++ b/README.md @@ -44,7 +44,7 @@ uv add "fastapi-toolsets[all]" ### Core -- **CRUD**: Generic async CRUD operations with `CrudFactory`, built-in full-text/faceted search and Offset/Cursor pagination. +- **CRUD**: Generic async CRUD operations with `CrudFactory`, built-in full-text/faceted search, Offset/Cursor pagination and bulk delete/update/custom actions. - **Database**: Session management, transaction helpers, table locking, and polling-based row change detection - **Dependencies**: FastAPI dependency factories (`PathDependency`, `BodyDependency`) for automatic DB lookups from path or body parameters - **Fixtures**: Fixture system with dependency management, context support, and pytest integration diff --git a/docs/module/crud.md b/docs/module/crud.md index 15764114..24f903d3 100644 --- a/docs/module/crud.md +++ b/docs/module/crud.md @@ -694,6 +694,88 @@ await UserCrud.upsert( ) ``` +## Bulk operations + +!!! info "Added in `v6.1`" + +`bulk_delete`, `bulk_update` and `bulk_apply` act on a list of primary keys, as sent by a UI where users tick rows and pick an action. The rows are loaded in one query, then changed through the ORM, so [lifecycle events](models.md), M2M fields and cascades behave as they do for `update` and `delete`. + +```python +from uuid import UUID + +from fastapi_toolsets.schemas import BulkRequest, BulkResult, BulkUpdateRequest, Response + + +@router.post("/bulk-delete") +async def bulk_delete_users( + session: SessionDep, body: BulkRequest[UUID] +) -> Response[BulkResult[UUID]]: + return await UserCrud.bulk_delete(session, body.ids, return_response=True) + + +@router.post("/bulk-update") +async def bulk_update_users( + session: SessionDep, body: BulkUpdateRequest[UUID, UserUpdate] +) -> Response[BulkResult[UUID]]: + return await UserCrud.bulk_update( + session, body.ids, body.data, return_response=True + ) +``` + +`bulk_update` applies the same patch to every row. As with `update`, only the fields set on the schema are written, and M2M id fields are resolved once for the whole batch. + +### Custom actions + +`bulk_apply` runs any async callable on each row. Raise an [`ApiException`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.ApiException) to fail a row: + +```python +async def archive(session: AsyncSession, user: User) -> None: + if user.is_admin: + raise ForbiddenError() + user.archived_at = datetime.now(UTC) + + +@router.post("/bulk-archive") +async def bulk_archive_users( + session: SessionDep, body: BulkRequest[UUID] +) -> Response[BulkResult[UUID]]: + return await UserCrud.bulk_apply( + session, body.ids, archive, mode="partial", return_response=True + ) +``` + +### Atomic and partial modes + +| Mode | Missing id | Failing row | Result | +| ---- | ---------- | ----------- | ------ | +| `"atomic"` (default) | raises [`NotFoundError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.NotFoundError) | the error propagates and every row rolls back | every id in `succeeded` | +| `"partial"` | reported in `failed` | its savepoint rolls back, the others are kept | `succeeded` and `failed` | + +In `"partial"` mode each row runs in its own savepoint. An `ApiException`, or a known constraint violation such as a foreign key still pointing at the row, fails only that row. Its error is reported in [`BulkResult.failed`](../reference/schemas.md#fastapi_toolsets.schemas.BulkResult): + +```json +{ + "status": "SUCCESS", + "data": { + "succeeded": ["3f1c..."], + "failed": [ + {"id": "9a0e...", "error_code": "AUTH-403", "message": "Forbidden", "description": "..."} + ] + } +} +``` + +Any other exception still aborts the whole batch. Lifecycle events fire only for the rows that were applied. + +!!! warning "A rolled-back row is expired." + SQLAlchemy expires the objects a rolled-back savepoint changed. Reading one of them afterwards reloads it, so keep the ids rather than the instances if you need them after the call. + +### Limits and locking + +- Ids must have the primary key's Python type (`BulkRequest[UUID]` parses them), with one tuple per row for a composite key. Duplicate ids are applied once. +- `max_items` (default `1000`) caps the number of distinct ids. Above it, [`BulkLimitExceededError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.BulkLimitExceededError) (HTTP 422) is raised before any query. Pass `max_items=None` to lift the cap. +- `with_for_update` locks the rows before the action runs. Rows are read in primary key order, so overlapping batches take their locks in the same order. + ## Response serialization !!! info "Added in `v1.1`" diff --git a/docs/module/exceptions.md b/docs/module/exceptions.md index 3e292985..c0a82bed 100644 --- a/docs/module/exceptions.md +++ b/docs/module/exceptions.md @@ -41,6 +41,7 @@ It also patches `app.openapi()` to replace the default Pydantic 422 schema with | [`InvalidFacetFilterError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidFacetFilterError) | 400 | Invalid Facet Filter | | [`InvalidFacetValueError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidFacetValueError) | 400 | Invalid Facet Value | | [`InvalidOrderFieldError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidOrderFieldError) | 422 | Invalid Order Field | +| [`BulkLimitExceededError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.BulkLimitExceededError) | 422 | Bulk Limit Exceeded | | [`PoolExhaustedError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.PoolExhaustedError) | 503 | Service Unavailable | | [`LockTimeoutError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.LockTimeoutError) | 503 | Service Unavailable | diff --git a/docs/reference/exceptions.md b/docs/reference/exceptions.md index 7018cd51..10d95243 100644 --- a/docs/reference/exceptions.md +++ b/docs/reference/exceptions.md @@ -16,6 +16,7 @@ from fastapi_toolsets.exceptions import ( InvalidFacetFilterError, InvalidFacetValueError, InvalidOrderFieldError, + BulkLimitExceededError, PoolExhaustedError, LockTimeoutError, IntegrityViolationError, @@ -50,6 +51,8 @@ from fastapi_toolsets.exceptions import ( ## ::: fastapi_toolsets.exceptions.exceptions.InvalidOrderFieldError +## ::: fastapi_toolsets.exceptions.exceptions.BulkLimitExceededError + ## ::: fastapi_toolsets.exceptions.exceptions.PoolExhaustedError ## ::: fastapi_toolsets.exceptions.exceptions.LockTimeoutError diff --git a/docs/reference/schemas.md b/docs/reference/schemas.md index d05b879a..31df9765 100644 --- a/docs/reference/schemas.md +++ b/docs/reference/schemas.md @@ -18,6 +18,10 @@ from fastapi_toolsets.schemas import ( PaginatedResponse, OffsetPaginatedResponse, CursorPaginatedResponse, + BulkRequest, + BulkUpdateRequest, + BulkItemError, + BulkResult, ) ``` @@ -44,3 +48,11 @@ from fastapi_toolsets.schemas import ( ## ::: fastapi_toolsets.schemas.OffsetPaginatedResponse ## ::: fastapi_toolsets.schemas.CursorPaginatedResponse + +## ::: fastapi_toolsets.schemas.BulkRequest + +## ::: fastapi_toolsets.schemas.BulkUpdateRequest + +## ::: fastapi_toolsets.schemas.BulkItemError + +## ::: fastapi_toolsets.schemas.BulkResult diff --git a/src/fastapi_toolsets/crud/factory.py b/src/fastapi_toolsets/crud/factory.py index 1d630def..ba43accb 100644 --- a/src/fastapi_toolsets/crud/factory.py +++ b/src/fastapi_toolsets/crud/factory.py @@ -31,7 +31,7 @@ tuple_, ) from sqlalchemy.dialects.postgresql import insert -from sqlalchemy.exc import NoResultFound +from sqlalchemy.exc import IntegrityError, NoResultFound from sqlalchemy.ext.asyncio import AsyncSession from sqlalchemy.orm import DeclarativeBase, QueryableAttribute, aliased, selectinload from sqlalchemy.orm.attributes import instance_state @@ -49,8 +49,17 @@ loaded_relationships, ) from ..db import transaction -from ..exceptions import InvalidOrderFieldError, NotFoundError +from ..exceptions import ( + ApiException, + BulkLimitExceededError, + InvalidOrderFieldError, + NotFoundError, + from_integrity_error, +) from ..schemas import ( + ApiError, + BulkItemError, + BulkResult, CursorPaginatedResponse, CursorPagination, OffsetPaginatedResponse, @@ -81,6 +90,8 @@ ) _ForUpdateMode: TypeAlias = bool | Literal["nowait", "skip_locked"] +_BulkMode: TypeAlias = Literal["atomic", "partial"] +_BulkAction: TypeAlias = Callable[[AsyncSession, Any], Awaitable[Any]] def _apply_for_update(q: Any, mode: _ForUpdateMode) -> Any: @@ -458,6 +469,21 @@ def _m2m_schema_fields(cls: type[Self]) -> set[str]: return set() return set(cls.m2m_fields.keys()) + @classmethod + def _m2m_load_options(cls: type[Self], obj: BaseModel) -> list[ExecutableOption]: + """Eager loads for the M2M collections *obj* sets. + + Assigning a collection that is not loaded would lazy-load it, which + fails in async. + """ + if not cls.m2m_fields: + return [] + return [ + selectinload(rel) + for schema_field, rel in cls.m2m_fields.items() + if schema_field in obj.model_fields_set + ] + @classmethod def _resolve_facet_fields( cls: type[Self], @@ -1444,20 +1470,11 @@ async def update( """ async with transaction(session): m2m_exclude = cls._m2m_schema_fields() - - # Eagerly load M2M relationships that will be updated so that - # setattr does not trigger a lazy load (which fails in async). - m2m_load_options: list[ExecutableOption] = [] - if m2m_exclude and cls.m2m_fields: - for schema_field, rel in cls.m2m_fields.items(): - if schema_field in obj.model_fields_set: - m2m_load_options.append(selectinload(rel)) - db_model = await cls.get( session=session, filters=filters, with_for_update=with_for_update, - load_options=m2m_load_options, + load_options=cls._m2m_load_options(obj), ) values = obj.model_dump( exclude_unset=exclude_unset, @@ -1570,6 +1587,347 @@ async def delete( return Response(data=None) return None + @staticmethod + def _bulk_ids(ids: Sequence[Any], max_items: int | None) -> list[Any]: + """*ids* without duplicates, checked against *max_items*.""" + unique = list(dict.fromkeys(ids)) + if max_items is not None and len(unique) > max_items: + raise BulkLimitExceededError(len(unique), max_items) + return unique + + @classmethod + async def _bulk_rows( + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + *, + with_for_update: _ForUpdateMode, + load_options: Sequence[ExecutableOption] | None, + ) -> dict[Any, ModelType]: + """Load the rows whose primary key is in *ids*, keyed by that primary key. + + Rows are read in primary key order, so concurrent calls that lock + overlapping sets take the locks in the same order. + """ + pk_attrs = cls._pk_attrs() + q = select(cls.model).where(key_expr(pk_attrs).in_(ids)).order_by(*pk_attrs) + if resolved := cls._resolve_load_options(load_options): + q = q.options(*resolved) + q = _apply_for_update(q, with_for_update) + result = await session.execute(q) + rows: dict[Any, ModelType] = {} + for obj in result.unique().scalars().all(): + identity = instance_state(obj).identity + assert identity is not None + rows[identity[0] if len(identity) == 1 else identity] = cast(ModelType, obj) + return rows + + @overload + @classmethod + async def bulk_apply( # pragma: no cover + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + action: _BulkAction, + *, + mode: _BulkMode = ..., + max_items: int | None = ..., + with_for_update: _ForUpdateMode = ..., + load_options: Sequence[ExecutableOption] | None = ..., + return_response: Literal[True], + ) -> Response[BulkResult[Any]]: ... + + @overload + @classmethod + async def bulk_apply( # pragma: no cover + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + action: _BulkAction, + *, + mode: _BulkMode = ..., + max_items: int | None = ..., + with_for_update: _ForUpdateMode = ..., + load_options: Sequence[ExecutableOption] | None = ..., + return_response: Literal[False] = ..., + ) -> BulkResult[Any]: ... + + @classmethod + async def bulk_apply( + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + action: _BulkAction, + *, + mode: _BulkMode = "atomic", + max_items: int | None = 1000, + with_for_update: _ForUpdateMode = False, + load_options: Sequence[ExecutableOption] | None = None, + return_response: bool = False, + ) -> BulkResult[Any] | Response[BulkResult[Any]]: + """Run *action* on every row whose primary key is in *ids*. + + The rows are loaded in one query, then ``await action(session, obj)`` + runs on each in the order of *ids*. Changes go through the ORM, so + lifecycle events, M2M handling and cascades behave as in ``update`` + and ``delete``. + + Args: + session: DB async session + ids: Primary keys of the rows, as the primary key's Python type + (a tuple per row for a composite key). Duplicates are ignored. + action: Async callable applied to each row. Raise an + ``ApiException`` to fail a row. + mode: ``"atomic"`` runs the whole batch in one transaction: a + missing id raises ``NotFoundError`` and the first error rolls + every row back. ``"partial"`` runs each row in its own + savepoint and reports missing ids, ``ApiException`` errors and + known constraint violations in ``BulkResult.failed`` instead; + any other error still aborts the batch. + max_items: Maximum number of distinct ids, or ``None`` for no limit. + with_for_update: Lock the rows before applying *action*. ``True`` + for plain ``FOR UPDATE``, ``"nowait"`` for ``FOR UPDATE NOWAIT``, + ``"skip_locked"`` for ``FOR UPDATE SKIP LOCKED``. + load_options: SQLAlchemy loader options for the rows, replacing + ``default_load_options``. + return_response: When ``True``, returns ``Response[BulkResult]``. + + Returns: + A ``BulkResult``, or ``Response[BulkResult]`` when + ``return_response=True``. + + Raises: + BulkLimitExceededError: If *ids* holds more than *max_items* ids + NotFoundError: In ``"atomic"`` mode, if an id matches no row + """ + unique = cls._bulk_ids(ids, max_items) + result: BulkResult[Any] = BulkResult() + if unique: + async with transaction(session): + rows = await cls._bulk_rows( + session, + unique, + with_for_update=with_for_update, + load_options=load_options, + ) + if mode == "atomic": + if missing := [i for i in unique if i not in rows]: + raise NotFoundError( + f"{cls.model.__name__} not found for IDs: " + + ", ".join(map(str, missing)) + ) + for id_ in unique: + await action(session, rows[id_]) + result.succeeded = unique + else: + for id_ in unique: + error = await cls._bulk_apply_one( + session, rows.get(id_), action + ) + if error is None: + result.succeeded.append(id_) + else: + result.failed.append( + BulkItemError( + id=id_, + error_code=error.err_code, + message=error.msg, + description=error.desc, + ) + ) + if return_response: + return Response(data=result) + return result + + @classmethod + async def _bulk_apply_one( + cls: type[Self], + session: AsyncSession, + obj: ModelType | None, + action: _BulkAction, + ) -> ApiError | None: + """Run *action* on *obj* in a savepoint; the error that rolled it back.""" + if obj is None: + return NotFoundError().api_error + try: + async with transaction(session): + await action(session, obj) + except ApiException as exc: + return exc.api_error + except IntegrityError as exc: + if (api_exc := from_integrity_error(exc)) is None: + raise + return api_exc.api_error + return None + + @overload + @classmethod + async def bulk_delete( # pragma: no cover + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + *, + mode: _BulkMode = ..., + max_items: int | None = ..., + with_for_update: _ForUpdateMode = ..., + return_response: Literal[True], + ) -> Response[BulkResult[Any]]: ... + + @overload + @classmethod + async def bulk_delete( # pragma: no cover + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + *, + mode: _BulkMode = ..., + max_items: int | None = ..., + with_for_update: _ForUpdateMode = ..., + return_response: Literal[False] = ..., + ) -> BulkResult[Any]: ... + + @classmethod + async def bulk_delete( + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + *, + mode: _BulkMode = "atomic", + max_items: int | None = 1000, + with_for_update: _ForUpdateMode = False, + return_response: bool = False, + ) -> BulkResult[Any] | Response[BulkResult[Any]]: + """Delete the rows whose primary key is in *ids*. + + Args: + session: DB async session + ids: Primary keys of the rows to delete + mode: ``"atomic"`` or ``"partial"``, see ``bulk_apply`` + max_items: Maximum number of distinct ids, or ``None`` for no limit + with_for_update: Lock the rows before deleting them + return_response: When ``True``, returns ``Response[BulkResult]``. + + Returns: + A ``BulkResult``, or ``Response[BulkResult]`` when + ``return_response=True``. + """ + + async def delete(session: AsyncSession, obj: Any) -> None: + await session.delete(obj) + + return await cls.bulk_apply( + session, + ids, + delete, + mode=mode, + max_items=max_items, + with_for_update=with_for_update, + load_options=[], + return_response=return_response, + ) + + @overload + @classmethod + async def bulk_update( # pragma: no cover + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + obj: BaseModel, + *, + exclude_unset: bool = ..., + exclude_none: bool = ..., + mode: _BulkMode = ..., + max_items: int | None = ..., + with_for_update: _ForUpdateMode = ..., + return_response: Literal[True], + ) -> Response[BulkResult[Any]]: ... + + @overload + @classmethod + async def bulk_update( # pragma: no cover + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + obj: BaseModel, + *, + exclude_unset: bool = ..., + exclude_none: bool = ..., + mode: _BulkMode = ..., + max_items: int | None = ..., + with_for_update: _ForUpdateMode = ..., + return_response: Literal[False] = ..., + ) -> BulkResult[Any]: ... + + @classmethod + async def bulk_update( + cls: type[Self], + session: AsyncSession, + ids: Sequence[Any], + obj: BaseModel, + *, + exclude_unset: bool = True, + exclude_none: bool = False, + mode: _BulkMode = "atomic", + max_items: int | None = 1000, + with_for_update: _ForUpdateMode = False, + return_response: bool = False, + ) -> BulkResult[Any] | Response[BulkResult[Any]]: + """Apply the same patch to the rows whose primary key is in *ids*. + + M2M id fields set on *obj* are resolved once and assigned to every row. + + Args: + session: DB async session + ids: Primary keys of the rows to update + obj: Pydantic model with update data + exclude_unset: Exclude fields not explicitly set in the schema + exclude_none: Exclude fields with None value + mode: ``"atomic"`` or ``"partial"``, see ``bulk_apply`` + max_items: Maximum number of distinct ids, or ``None`` for no limit + with_for_update: Lock the rows before updating them + return_response: When ``True``, returns ``Response[BulkResult]``. + + Returns: + A ``BulkResult``, or ``Response[BulkResult]`` when + ``return_response=True``. + + Raises: + NotFoundError: If an M2M id matches no related row + """ + unique = cls._bulk_ids(ids, max_items) + m2m_exclude = cls._m2m_schema_fields() + values = obj.model_dump( + exclude_unset=exclude_unset, + exclude_none=exclude_none, + exclude=m2m_exclude, + ) + + # The outer transaction covers resolving the M2M ids, so an unknown + # id rolls back like any other failure. + async with transaction(session): + m2m_resolved = ( + await cls._resolve_m2m(session, obj, only_set=True) + if m2m_exclude + else {} + ) + + async def update(session: AsyncSession, db_model: Any) -> None: + for key, value in values.items(): + setattr(db_model, key, value) + for rel_attr, related_instances in m2m_resolved.items(): + setattr(db_model, rel_attr, list(related_instances)) + + return await cls.bulk_apply( + session, + unique, + update, + mode=mode, + max_items=max_items, + with_for_update=with_for_update, + load_options=cls._m2m_load_options(obj), + return_response=return_response, + ) + @classmethod async def count( cls: type[Self], diff --git a/src/fastapi_toolsets/exceptions/__init__.py b/src/fastapi_toolsets/exceptions/__init__.py index 8f1de41e..fac0a449 100644 --- a/src/fastapi_toolsets/exceptions/__init__.py +++ b/src/fastapi_toolsets/exceptions/__init__.py @@ -3,6 +3,7 @@ from .exceptions import ( ApiError, ApiException, + BulkLimitExceededError, CheckViolationError, ConflictError, ExclusionViolationError, @@ -29,6 +30,7 @@ __all__ = [ "ApiError", "ApiException", + "BulkLimitExceededError", "CheckViolationError", "ConflictError", "ExclusionViolationError", diff --git a/src/fastapi_toolsets/exceptions/exceptions.py b/src/fastapi_toolsets/exceptions/exceptions.py index e8f45b84..4bf80966 100644 --- a/src/fastapi_toolsets/exceptions/exceptions.py +++ b/src/fastapi_toolsets/exceptions/exceptions.py @@ -247,6 +247,30 @@ def __init__(self, field: str, valid_fields: list[str]) -> None: ) +class BulkLimitExceededError(ApiException): + """Raised when a bulk operation targets more rows than allowed.""" + + api_error = ApiError( + code=422, + msg="Bulk Limit Exceeded", + desc="The bulk operation targets too many rows.", + err_code="BULK-422", + ) + + def __init__(self, count: int, limit: int) -> None: + """Initialize the exception. + + Args: + count: Number of distinct ids the caller sent. + limit: Maximum number of ids one call accepts. + """ + self.count = count + self.limit = limit + super().__init__( + desc=f"{count} ids were given, but at most {limit} are allowed per call." + ) + + class PoolExhaustedError(ApiException): """HTTP 503 - Database connection pool is exhausted.""" diff --git a/src/fastapi_toolsets/schemas.py b/src/fastapi_toolsets/schemas.py index d0919985..583ac40e 100644 --- a/src/fastapi_toolsets/schemas.py +++ b/src/fastapi_toolsets/schemas.py @@ -6,10 +6,14 @@ from pydantic import BaseModel, ConfigDict, Field, computed_field -from .types import DataT +from .types import DataT, IdT, SchemaType __all__ = [ "ApiError", + "BulkItemError", + "BulkRequest", + "BulkResult", + "BulkUpdateRequest", "CursorPaginatedResponse", "CursorPagination", "ErrorResponse", @@ -204,3 +208,54 @@ class CursorPaginatedResponse(PaginatedResponse[DataT]): pagination: CursorPagination pagination_type: Literal[PaginationType.CURSOR] = PaginationType.CURSOR + + +class BulkRequest(PydanticBase, Generic[IdT]): + """Request body selecting the rows a bulk operation acts on. + + Example: + ```python + @router.post("/bulk-delete") + async def bulk_delete( + session: SessionDep, body: BulkRequest[UUID] + ) -> Response[BulkResult[UUID]]: + return await UserCrud.bulk_delete(session, body.ids, return_response=True) + ``` + """ + + ids: list[IdT] = Field(min_length=1) + + +class BulkUpdateRequest(BulkRequest[IdT], Generic[IdT, SchemaType]): + """Request body applying one patch to every selected row.""" + + data: SchemaType + + +class BulkItemError(PydanticBase, Generic[IdT]): + """Why a bulk operation failed for one row. + + Attributes: + id: Primary key of the row + error_code: Application-specific error code (e.g., "RES-404") + message: Short error message + description: Detailed error description + """ + + id: IdT + error_code: str + message: str + description: str | None = None + + +class BulkResult(PydanticBase, Generic[IdT]): + """Outcome of a bulk operation. + + Attributes: + succeeded: Primary keys of the rows the operation was applied to + failed: One entry per row it could not be applied to. Always empty + in ``"atomic"`` mode, where the first failure aborts the batch. + """ + + succeeded: list[IdT] = Field(default_factory=list) + failed: list[BulkItemError[IdT]] = Field(default_factory=list) diff --git a/src/fastapi_toolsets/types.py b/src/fastapi_toolsets/types.py index 517a3e55..063ceeb0 100644 --- a/src/fastapi_toolsets/types.py +++ b/src/fastapi_toolsets/types.py @@ -11,6 +11,7 @@ # Generic TypeVars DataT = TypeVar("DataT") +IdT = TypeVar("IdT") ModelType = TypeVar("ModelType", bound=DeclarativeBase) SchemaType = TypeVar("SchemaType", bound=BaseModel) diff --git a/tests/test_crud_bulk.py b/tests/test_crud_bulk.py new file mode 100644 index 00000000..a0fc73ab --- /dev/null +++ b/tests/test_crud_bulk.py @@ -0,0 +1,385 @@ +"""Tests for the bulk operations of ``AsyncCrud``.""" + +import uuid +from typing import Any + +import pytest +from sqlalchemy import select +from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker + +from fastapi_toolsets.crud import CrudFactory +from fastapi_toolsets.crud import factory as crud_factory +from fastapi_toolsets.exceptions import ( + BulkLimitExceededError, + ForbiddenError, + NotFoundError, +) +from fastapi_toolsets.models import EventSession, ModelEvent, listens_for +from fastapi_toolsets.models.watched import _EVENT_HANDLERS, _invalidate_caches +from fastapi_toolsets.schemas import ( + BulkItemError, + BulkRequest, + BulkResult, + BulkUpdateRequest, + Response, +) + +from .conftest import ( + Permission, + Post, + PostCreate, + PostCrud, + PostM2MUpdate, + PostTagsLoadCrud, + Role, + RoleCrud, + RoleUpdate, + Tag, + TagCreate, + TagCrud, + User, + UserCrud, + UserUpdate, + capture_sql, + create_role, + create_user, + created_tables, +) + +PermissionCrud = CrudFactory(Permission) + + +async def _names(session: AsyncSession) -> list[str]: + return list(await session.scalars(select(Role.name).order_by(Role.name))) + + +async def _forbid_admin(session: AsyncSession, role: Role) -> None: + if role.name == "admin": + raise ForbiddenError() + role.name = f"{role.name}-done" + + +class TestSchemas: + """Request and result models.""" + + def test_requests_parse_ids_and_patch(self): + role_id = uuid.uuid4() + + body = BulkUpdateRequest[uuid.UUID, RoleUpdate].model_validate( + {"ids": [str(role_id)], "data": {"name": "x"}} + ) + + assert body.ids == [role_id] and body.data.name == "x" + with pytest.raises(ValueError): + BulkRequest[uuid.UUID](ids=[]) + + def test_result_defaults_to_empty_lists(self): + assert BulkResult[int]().model_dump() == {"succeeded": [], "failed": []} + + +class TestBulkApply: + """bulk_apply(): selection, modes, limits and locking.""" + + @pytest.mark.anyio + async def test_atomic_applies_the_action_once_per_distinct_id(self, db_session): + a, b = await create_role(db_session, "a"), await create_role(db_session, "b") + seen: list[uuid.UUID] = [] + + async def action(session: AsyncSession, role: Role) -> None: + seen.append(role.id) + role.name = role.name.upper() + + result = await RoleCrud.bulk_apply(db_session, [b.id, a.id, b.id], action) + + assert result == BulkResult(succeeded=[b.id, a.id]) + assert seen == [b.id, a.id] + assert await _names(db_session) == ["A", "B"] + + @pytest.mark.anyio + async def test_atomic_rolls_back_every_row_on_a_missing_id(self, db_session): + role = await create_role(db_session, "a") + missing = uuid.uuid4() + + with pytest.raises(NotFoundError, match=f"Role not found for IDs: {missing}"): + await RoleCrud.bulk_apply(db_session, [role.id, missing], _forbid_admin) + + db_session.expire_all() + assert await _names(db_session) == ["a"] + + @pytest.mark.anyio + async def test_atomic_rolls_back_every_row_on_an_error(self, db_session): + user = await create_role(db_session, "user") + admin = await create_role(db_session, "admin") + + with pytest.raises(ForbiddenError): + await RoleCrud.bulk_apply(db_session, [user.id, admin.id], _forbid_admin) + + db_session.expire_all() + assert await _names(db_session) == ["admin", "user"] + + @pytest.mark.anyio + async def test_partial_reports_failures_and_keeps_the_rest(self, db_session): + user = await create_role(db_session, "user") + admin = await create_role(db_session, "admin") + missing = uuid.uuid4() + + response = await RoleCrud.bulk_apply( + db_session, + [admin.id, missing, user.id], + _forbid_admin, + mode="partial", + return_response=True, + ) + + assert isinstance(response, Response) + assert response.data == BulkResult( + succeeded=[user.id], + failed=[ + BulkItemError( + id=admin.id, + error_code="AUTH-403", + message="Forbidden", + description=ForbiddenError.api_error.desc, + ), + BulkItemError( + id=missing, + error_code="RES-404", + message="Not Found", + description=NotFoundError.api_error.desc, + ), + ], + ) + db_session.expire_all() + assert await _names(db_session) == ["admin", "user-done"] + + @pytest.mark.anyio + async def test_the_limit_counts_distinct_ids(self, db_session): + a, b = await create_role(db_session, "a"), await create_role(db_session, "b") + + with pytest.raises(BulkLimitExceededError, match="Bulk Limit Exceeded"): + await RoleCrud.bulk_apply( + db_session, [a.id, b.id], _forbid_admin, max_items=1 + ) + same = await RoleCrud.bulk_apply( + db_session, [a.id, a.id], _forbid_admin, max_items=1 + ) + unlimited = await RoleCrud.bulk_apply( + db_session, [a.id, b.id], _forbid_admin, max_items=None + ) + + assert same.succeeded == [a.id] and unlimited.succeeded == [a.id, b.id] + + @pytest.mark.anyio + async def test_no_ids_sends_no_query(self, engine, db_session): + with capture_sql(engine) as statements: + result = await RoleCrud.bulk_apply(db_session, [], _forbid_admin) + + assert result == BulkResult() and statements == [] + + @pytest.mark.anyio + async def test_rows_load_in_one_query_in_key_order_and_can_be_locked( + self, engine, db_session + ): + a, b = await create_role(db_session, "a"), await create_role(db_session, "b") + + with capture_sql(engine) as statements: + await RoleCrud.bulk_apply( + db_session, [a.id, b.id], _forbid_admin, with_for_update="nowait" + ) + + [load] = [sql for sql in statements if sql.startswith("SELECT")] + assert "ORDER BY roles.id" in load and load.endswith("FOR UPDATE NOWAIT") + + @pytest.mark.anyio + async def test_inside_an_open_transaction_only_the_batch_rolls_back( + self, db_session + ): + admin = await create_role(db_session, "admin") + + async with db_session.begin(): + await create_role(db_session, "kept") + with pytest.raises(ForbiddenError): + await RoleCrud.bulk_apply(db_session, [admin.id], _forbid_admin) + + assert await _names(db_session) == ["admin", "kept"] + + +class TestBulkDelete: + """bulk_delete().""" + + @pytest.mark.anyio + async def test_deletes_the_rows(self, db_session): + a, b = await create_role(db_session, "a"), await create_role(db_session, "b") + await create_role(db_session, "c") + + result = await RoleCrud.bulk_delete(db_session, [a.id, b.id]) + + assert result.succeeded == [a.id, b.id] + assert await _names(db_session) == ["c"] + + @pytest.mark.anyio + async def test_partial_reports_constraint_violations(self, db_session): + author = await create_user(db_session, "author") + idle = await create_user(db_session, "idle") + await PostCrud.create(db_session, PostCreate(title="t", author_id=author.id)) + + response = await UserCrud.bulk_delete( + db_session, [author.id, idle.id], mode="partial", return_response=True + ) + + assert isinstance(response, Response) and response.data is not None + assert response.data.succeeded == [idle.id] + assert [(f.id, f.error_code) for f in response.data.failed] == [ + (author.id, "DB-409-FK") + ] + assert [u.id for u in await UserCrud.get_multi(db_session)] == [author.id] + + @pytest.mark.anyio + async def test_partial_raises_an_unknown_constraint_violation( + self, db_session, monkeypatch + ): + author = await create_user(db_session, "author") + await PostCrud.create(db_session, PostCreate(title="t", author_id=author.id)) + monkeypatch.setattr(crud_factory, "from_integrity_error", lambda exc: None) + + with pytest.raises(Exception, match="ForeignKeyViolationError"): + await UserCrud.bulk_delete(db_session, [author.id], mode="partial") + + @pytest.mark.anyio + async def test_composite_keys_are_given_as_tuples(self, db_session): + db_session.add_all( + [Permission(subject="post", action=a) for a in ("read", "write")] + ) + await db_session.commit() + + result = await PermissionCrud.bulk_delete( + db_session, [("post", "write"), ("post", "nope")], mode="partial" + ) + + assert result.succeeded == [("post", "write")] + assert [f.id for f in result.failed] == [("post", "nope")] + assert [p.action for p in await PermissionCrud.get_multi(db_session)] == [ + "read" + ] + + +class TestBulkUpdate: + """bulk_update().""" + + @pytest.mark.anyio + async def test_applies_the_set_fields_to_every_row(self, db_session): + alice = await create_user(db_session, "alice") + bob = await create_user(db_session, "bob") + + response = await UserCrud.bulk_update( + db_session, + [alice.id, bob.id], + UserUpdate(is_active=False), + return_response=True, + ) + + assert isinstance(response, Response) and response.data is not None + assert response.data.succeeded == [alice.id, bob.id] + db_session.expire_all() + users = await UserCrud.get_multi(db_session, order_by=User.username) + assert [(u.username, u.is_active) for u in users] == [ + ("alice", False), + ("bob", False), + ] + + @pytest.mark.anyio + async def test_partial_reports_constraint_violations(self, db_session): + # A rolled-back savepoint expires the rows it changed: keep plain ids. + alice = (await create_user(db_session, "alice")).id + bob = (await create_user(db_session, "bob")).id + + result = await UserCrud.bulk_update( + db_session, [alice, bob], UserUpdate(email="same"), mode="partial" + ) + + assert result.succeeded == [alice] + assert [(f.id, f.error_code) for f in result.failed] == [(bob, "DB-409-UNIQUE")] + + @pytest.mark.anyio + async def test_m2m_ids_are_resolved_once_for_every_row(self, engine, db_session): + author = await create_user(db_session, "author") + posts = [ + await PostCrud.create(db_session, PostCreate(title=t, author_id=author.id)) + for t in ("a", "b") + ] + tag = await TagCrud.create(db_session, TagCreate(name="news")) + + with capture_sql(engine) as statements: + await PostTagsLoadCrud.bulk_update( + db_session, [p.id for p in posts], PostM2MUpdate(tag_ids=[tag.id]) + ) + + assert sum("WHERE tags.id IN" in sql for sql in statements) == 1 + rows = await PostTagsLoadCrud.get_multi(db_session, order_by=Post.title) + assert [[t.id for t in p.tags] for p in rows] == [[tag.id], [tag.id]] + + @pytest.mark.anyio + async def test_the_limit_is_checked_before_resolving_m2m_ids( + self, engine, db_session + ): + with capture_sql(engine) as statements: + with pytest.raises(BulkLimitExceededError): + await PostTagsLoadCrud.bulk_update( + db_session, + [uuid.uuid4(), uuid.uuid4()], + PostM2MUpdate(tag_ids=[uuid.uuid4()]), + max_items=1, + ) + + assert statements == [] + + @pytest.mark.anyio + async def test_an_unknown_m2m_id_rolls_back_in_any_mode(self, db_session): + author = await create_user(db_session, "author") + post = await PostCrud.create( + db_session, PostCreate(title="a", author_id=author.id) + ) + + with pytest.raises(NotFoundError, match="Related Tag not found"): + await PostTagsLoadCrud.bulk_update( + db_session, + [post.id], + PostM2MUpdate(title="x", tag_ids=[uuid.uuid4()]), + mode="partial", + ) + + assert await db_session.scalar(select(Post.title)) == "a" + assert await db_session.scalar(select(Tag.id)) is None + + +class TestEvents: + """Lifecycle events under an ``EventSession``.""" + + @pytest.fixture + async def event_session(self, engine): + async with created_tables(engine): + maker = async_sessionmaker( + engine, expire_on_commit=False, class_=EventSession + ) + async with maker() as session: + yield session + for key in list(_EVENT_HANDLERS): + if key[0] is Role: + del _EVENT_HANDLERS[key] + _invalidate_caches() + + @pytest.mark.anyio + async def test_only_the_applied_rows_dispatch(self, event_session): + user = await create_role(event_session, "user") + admin = await create_role(event_session, "admin") + seen: list[tuple[str, Any]] = [] + + @listens_for(Role, [ModelEvent.UPDATE]) + async def _on_update(obj: Role, event_type: ModelEvent, changes: Any) -> None: + seen.append((obj.name, changes["name"]["old"])) + + await RoleCrud.bulk_apply( + event_session, [admin.id, user.id], _forbid_admin, mode="partial" + ) + + assert seen == [("user-done", "user")]