refactor: simplify and deduplicate across crud, metrics, cli, and

exceptions
test: add missing tests for fixtures/utils.py
2026-08-14 11:57:01 +00:00 · 2026-03-01 09:47:36 -05:00 · 2026-03-01 08:05:20 -05:00 · 2026-03-01 07:46:49 -05:00 · 2026-03-01 07:21:07 -05:00
21 changed files with 199 additions and 1285 deletions
@@ -20,7 +20,7 @@ A modular collection of production-ready utilities for FastAPI. Install only wha
 ## Installation
-The base package includes the core modules (CRUD, database, schemas, exceptions, fixtures, dependencies, model mixins, logging):
+The base package includes the core modules (CRUD, database, schemas, exceptions, fixtures, dependencies, logging):
 ```bash
 uv add fastapi-toolsets
@@ -29,9 +29,9 @@ uv add fastapi-toolsets
 Install only the extras you need:
 ```bash
-uv add "fastapi-toolsets[cli]"
+uv add "fastapi-toolsets[cli]"      # CLI (typer)
-uv add "fastapi-toolsets[metrics]"
+uv add "fastapi-toolsets[metrics]"  # Prometheus metrics (prometheus_client)
-uv add "fastapi-toolsets[pytest]"
+uv add "fastapi-toolsets[pytest]"   # Pytest helpers (httpx, pytest-xdist)
 ```
 Or install everything:
@@ -48,7 +48,6 @@ uv add "fastapi-toolsets[all]"
 - **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
 - **Model Mixins**: SQLAlchemy mixins for common column patterns (`UUIDMixin`, `CreatedAtMixin`, `UpdatedAtMixin`, `TimestampMixin`)
 - **Standardized API Responses**: Consistent response format with `Response`, `PaginatedResponse`, and `PydanticBase`
 - **Exception Handling**: Structured error responses with automatic OpenAPI documentation
 - **Logging**: Logging configuration with uvicorn integration via `configure_logging` and `get_logger`
@@ -20,7 +20,7 @@ A modular collection of production-ready utilities for FastAPI. Install only wha
 ## Installation
-The base package includes the core modules (CRUD, database, schemas, exceptions, fixtures, dependencies, model mixins, logging):
+The base package includes the core modules (CRUD, database, schemas, exceptions, fixtures, dependencies, logging):
 ```bash
 uv add fastapi-toolsets
@@ -29,9 +29,9 @@ uv add fastapi-toolsets
 Install only the extras you need:
 ```bash
-uv add "fastapi-toolsets[cli]"
+uv add "fastapi-toolsets[cli]"      # CLI (typer)
-uv add "fastapi-toolsets[metrics]"
+uv add "fastapi-toolsets[metrics]"  # Prometheus metrics (prometheus_client)
-uv add "fastapi-toolsets[pytest]"
+uv add "fastapi-toolsets[pytest]"   # Pytest helpers (httpx, pytest-xdist)
 ```
 Or install everything:
@@ -44,11 +44,10 @@ 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 and offset/cursor pagination.
 - **Database**: Session management, transaction helpers, table locking, and polling-based row change detection
 - **Dependencies**: FastAPI dependency factories (`PathDependency`, `BodyDependency`) for automatic DB lookups from path or body parameters
 - **Fixtures**: Fixture system with dependency management, context support, and pytest integration
 - **Model Mixins**: SQLAlchemy mixins for common column patterns (`UUIDMixin`, `CreatedAtMixin`, `UpdatedAtMixin`, `TimestampMixin`)
 - **Standardized API Responses**: Consistent response format with `Response`, `PaginatedResponse`, and `PydanticBase`
 - **Exception Handling**: Structured error responses with automatic OpenAPI documentation
 - **Logging**: Logging configuration with uvicorn integration via `configure_logging` and `get_logger`
@@ -1,137 +0,0 @@
 # Migrating to v2.0
 This page covers every breaking change introduced in **v2.0** and the steps required to update your code.
 ---
 ## CRUD
 ### `schema` is now required in `offset_paginate()` and `cursor_paginate()`
 Calls that omit `schema` will now raise a `TypeError` at runtime.
 Previously `schema` was optional; omitting it returned raw SQLAlchemy model instances inside the response. It is now a required keyword argument and the response always contains serialized schema instances.
 === "Before (`v1`)"
    ```python
    # schema omitted — returned raw model instances
    result = await UserCrud.offset_paginate(session=session, page=1)
    result = await UserCrud.cursor_paginate(session=session, cursor=token)
    ```
 === "Now (`v2`)"
    ```python
    result = await UserCrud.offset_paginate(session=session, page=1, schema=UserRead)
    result = await UserCrud.cursor_paginate(session=session, cursor=token, schema=UserRead)
    ```
 ### `as_response` removed from `create()`, `get()`, and `update()`
 Passing `as_response` to these methods will raise a `TypeError` at runtime.
 The `as_response=True` shorthand is replaced by passing a `schema` directly. The return value is a `Response[schema]` when `schema` is provided, or the raw model instance when it is not.
 === "Before (`v1`)"
    ```python
    user = await UserCrud.create(session=session, obj=data, as_response=True)
    user = await UserCrud.get(session=session, filters=filters, as_response=True)
    user = await UserCrud.update(session=session, obj=data, filters, as_response=True)
    ```
 === "Now (`v2`)"
    ```python
    user = await UserCrud.create(session=session, obj=data, schema=UserRead)
    user = await UserCrud.get(session=session, filters=filters, schema=UserRead)
    user = await UserCrud.update(session=session, obj=data, filters, schema=UserRead)
    ```
 ### `delete()`: `as_response` renamed and return type changed
 `as_response` is gone, and the plain (non-response) call no longer returns `True`.
 Two changes were made to `delete()`:
 1. The `as_response` parameter is renamed to `return_response`.
 2. When called without `return_response=True`, the method now returns `None` on success instead of `True`.
 === "Before (`v1`)"
    ```python
    ok = await UserCrud.delete(session=session, filters=filters)
    if ok:  # True on success
        ...
    response = await UserCrud.delete(session=session, filters=filters, as_response=True)
    ```
 === "Now (`v2`)"
    ```python
    await UserCrud.delete(session=session, filters=filters)  # returns None
    response = await UserCrud.delete(session=session, filters=filters, return_response=True)
    ```
 ### `paginate()` alias removed
 Any call to `crud.paginate(...)` will raise `AttributeError` at runtime.
 The `paginate` shorthand was an alias for `offset_paginate`. It has been removed; call `offset_paginate` directly.
 === "Before (`v1`)"
    ```python
    result = await UserCrud.paginate(session=session, page=2, items_per_page=20, schema=UserRead)
    ```
 === "Now (`v2`)"
    ```python
    result = await UserCrud.offset_paginate(session=session, page=2, items_per_page=20, schema=UserRead)
    ```
 ---
 ## Exceptions
 ### Missing `api_error` raises `TypeError` at class definition time
 Unfinished or stub exception subclasses that previously compiled fine will now fail on import.
 In `v1`, a subclass without `api_error` would only fail when the exception was raised. In `v2`, `__init_subclass__` validates this at class definition time.
 === "Before (`v1`)"
    ```python
    class MyError(ApiException):
        pass  # fine until raised
    ```
 === "Now (`v2`)"
    ```python
    class MyError(ApiException):
        pass  # TypeError: MyError must define an 'api_error' class attribute.
    ```
 For shared base classes that are not meant to be raised directly, use `abstract=True`:
 ```python
 class BillingError(ApiException, abstract=True):
    """Base for all billing-related errors — not raised directly."""
 class PaymentRequiredError(BillingError):
    api_error = ApiError(code=402, msg="Payment Required", desc="...", err_code="BILLING-402")
 ```
 ---
 ## Schemas
 ### `Pagination` alias removed
 `Pagination` was already deprecated in `v1` and is fully removed in `v2`, you now need to use [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination) or [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination).
@@ -21,37 +21,30 @@ init_exceptions_handlers(app=app)
 This registers handlers for:
 - [`ApiException`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.ApiException) — all custom exceptions below
 - `HTTPException` — Starlette/FastAPI HTTP errors
 - `RequestValidationError` — Pydantic request validation (422)
 - `ResponseValidationError` — Pydantic response validation (422)
 - `Exception` — unhandled errors (500)
 It also patches `app.openapi()` to replace the default Pydantic 422 schema with a structured example matching the `ErrorResponse` format.
 ## Built-in exceptions
 | Exception | Status | Default message |
 |-----------|--------|-----------------|
 | [`UnauthorizedError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.UnauthorizedError) | 401 | Unauthorized |
 | [`ForbiddenError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.ForbiddenError) | 403 | Forbidden |
-| [`NotFoundError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.NotFoundError) | 404 | Not Found |
+| [`NotFoundError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.NotFoundError) | 404 | Not found |
 | [`ConflictError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.ConflictError) | 409 | Conflict |
-| [`NoSearchableFieldsError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.NoSearchableFieldsError) | 400 | No Searchable Fields |
+| [`NoSearchableFieldsError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.NoSearchableFieldsError) | 400 | No searchable fields |
-| [`InvalidFacetFilterError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidFacetFilterError) | 400 | Invalid Facet Filter |
+| [`InvalidFacetFilterError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidFacetFilterError) | 400 | Invalid facet filter |
 | [`InvalidOrderFieldError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidOrderFieldError) | 422 | Invalid Order Field |
 ### Per-instance overrides
 All built-in exceptions accept optional keyword arguments to customise the response for a specific raise site without changing the class defaults:
 | Argument | Effect |
 |----------|--------|
 | `detail` | Overrides both `str(exc)` (log output) and the `message` field in the response body |
 | `desc` | Overrides the `description` field |
 | `data` | Overrides the `data` field |
 ```python
-raise NotFoundError(detail="User 42 not found", desc="No user with that ID exists in the database.")
+from fastapi_toolsets.exceptions import NotFoundError
@router.get("/users/{id}")
 async def get_user(id: int, session: AsyncSession = Depends(get_db)):
    user = await UserCrud.first(session=session, filters=[User.id == id])
    if not user:
        raise NotFoundError
    return user
 ```
 ## Custom exceptions
@@ -65,51 +58,12 @@ from fastapi_toolsets.schemas import ApiError
 class PaymentRequiredError(ApiException):
    api_error = ApiError(
        code=402,
-        msg="Payment Required",
+        msg="Payment required",
        desc="Your subscription has expired.",
-        err_code="BILLING-402",
+        err_code="PAYMENT_REQUIRED",
    )
 ```
 !!! warning
    Subclasses that do not define `api_error` raise a `TypeError` at **class creation time**, not at raise time.
 ### Custom `__init__`
 Override `__init__` to compute `detail`, `desc`, or `data` dynamically, then delegate to `super().__init__()`:
 ```python
 class OrderValidationError(ApiException):
    api_error = ApiError(
        code=422,
        msg="Order Validation Failed",
        desc="One or more order fields are invalid.",
        err_code="ORDER-422",
    )
    def __init__(self, *field_errors: str) -> None:
        super().__init__(
            f"{len(field_errors)} validation error(s)",
            desc=", ".join(field_errors),
            data={"errors": [{"message": e} for e in field_errors]},
        )
 ```
 ### Intermediate base classes
 Use `abstract=True` when creating a shared base that is not meant to be raised directly:
 ```python
 class BillingError(ApiException, abstract=True):
    """Base for all billing-related errors."""
 class PaymentRequiredError(BillingError):
    api_error = ApiError(code=402, msg="Payment Required", desc="...", err_code="BILLING-402")
 class SubscriptionExpiredError(BillingError):
    api_error = ApiError(code=402, msg="Subscription Expired", desc="...", err_code="BILLING-402-EXP")
 ```
 ## OpenAPI response documentation
 Use [`generate_error_responses`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.generate_error_responses) to add error schemas to your endpoint's OpenAPI spec:
@@ -124,7 +78,8 @@ from fastapi_toolsets.exceptions import generate_error_responses, NotFoundError,
 async def get_user(...): ...
 ```
-Multiple exceptions sharing the same HTTP status code are grouped under one entry, each appearing as a named example keyed by its `err_code`. This keeps the OpenAPI UI readable when several error variants map to the same status.
+!!! info
    The pydantic validation error is automatically added by FastAPI.
 ---
@@ -1,113 +0,0 @@
 # Models
 !!! info "Added in `v2.0`"
 Reusable SQLAlchemy 2.0 mixins for common column patterns, designed to be composed freely on any `DeclarativeBase` model.
 ## Overview
 The `models` module provides mixins that each add a single, well-defined column behaviour. They work with standard SQLAlchemy 2.0 declarative syntax and are fully compatible with `AsyncSession`.
 ```python
 from fastapi_toolsets.models import UUIDMixin, TimestampMixin
 class Article(Base, UUIDMixin, TimestampMixin):
    __tablename__ = "articles"
    title: Mapped[str]
    content: Mapped[str]
 ```
 All timestamp columns are timezone-aware (`TIMESTAMPTZ`). All defaults are server-side, so they are also applied when inserting rows via raw SQL outside the ORM.
 ## Mixins
 ### [`UUIDMixin`](../reference/models.md#fastapi_toolsets.models.UUIDMixin)
 Adds a `id: UUID` primary key generated server-side by PostgreSQL using `gen_random_uuid()` (requires PostgreSQL 13+). The value is retrieved via `RETURNING` after insert, so it is available on the Python object immediately after `flush()`.
 ```python
 from fastapi_toolsets.models import UUIDMixin
 class User(Base, UUIDMixin):
    __tablename__ = "users"
    username: Mapped[str]
 # id is None before flush
 user = User(username="alice")
 await session.flush()
 print(user.id)  # UUID('...')
 ```
 ### [`CreatedAtMixin`](../reference/models.md#fastapi_toolsets.models.CreatedAtMixin)
 Adds a `created_at: datetime` column set to `NOW()` on insert. The column has no `onupdate` hook — it is intentionally immutable after the row is created.
 ```python
 from fastapi_toolsets.models import UUIDMixin, CreatedAtMixin
 class Order(Base, UUIDMixin, CreatedAtMixin):
    __tablename__ = "orders"
    total: Mapped[float]
 ```
 ### [`UpdatedAtMixin`](../reference/models.md#fastapi_toolsets.models.UpdatedAtMixin)
 Adds an `updated_at: datetime` column set to `NOW()` on insert and automatically updated to `NOW()` on every ORM-level update (via SQLAlchemy's `onupdate` hook).
 ```python
 from fastapi_toolsets.models import UUIDMixin, UpdatedAtMixin
 class Post(Base, UUIDMixin, UpdatedAtMixin):
    __tablename__ = "posts"
    title: Mapped[str]
 post = Post(title="Hello")
 await session.flush()
 await session.refresh(post)
 post.title = "Hello World"
 await session.flush()
 await session.refresh(post)
 print(post.updated_at)
 ```
 !!! note
    `updated_at` is updated by SQLAlchemy at ORM flush time. If you update rows via raw SQL (e.g. `UPDATE posts SET ...`), the column will **not** be updated automatically — use a database trigger if you need that guarantee.
 ### [`TimestampMixin`](../reference/models.md#fastapi_toolsets.models.TimestampMixin)
 Convenience mixin that combines [`CreatedAtMixin`](../reference/models.md#fastapi_toolsets.models.CreatedAtMixin) and [`UpdatedAtMixin`](../reference/models.md#fastapi_toolsets.models.UpdatedAtMixin). Equivalent to inheriting both.
 ```python
 from fastapi_toolsets.models import UUIDMixin, TimestampMixin
 class Article(Base, UUIDMixin, TimestampMixin):
    __tablename__ = "articles"
    title: Mapped[str]
 ```
 ## Composing mixins
 All mixins can be combined in any order. The only constraint is that exactly one primary key must be defined — either via `UUIDMixin` or directly on the model.
 ```python
 from fastapi_toolsets.models import UUIDMixin, TimestampMixin
 class Event(Base, UUIDMixin, TimestampMixin):
    __tablename__ = "events"
    name: Mapped[str]
 class Counter(Base, UpdatedAtMixin):
    __tablename__ = "counters"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    value: Mapped[int]
 ```
 ---
 [:material-api: API Reference](../reference/models.md)
@@ -22,20 +22,16 @@ async def get_user(user: User = UserDep) -> Response[UserSchema]:
 ### [`PaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse)
-Wraps a list of items with pagination metadata and optional facet values. The `pagination` field accepts either [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination) or [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination) depending on the strategy used.
+Wraps a list of items with pagination metadata and optional facet values.
 #### [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination)
 Page-number based. Requires `total_count` so clients can compute the total number of pages.
 ```python
-from fastapi_toolsets.schemas import PaginatedResponse, OffsetPagination
+from fastapi_toolsets.schemas import PaginatedResponse, Pagination
@router.get("/users")
 async def list_users() -> PaginatedResponse[UserSchema]:
    return PaginatedResponse(
        data=users,
-        pagination=OffsetPagination(
+        pagination=Pagination(
            total_count=100,
            items_per_page=10,
            page=1,
@@ -44,26 +40,6 @@ async def list_users() -> PaginatedResponse[UserSchema]:
    )
 ```
 #### [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination)
 Cursor based. Efficient for large or frequently updated datasets where offset pagination is impractical. Provides opaque `next_cursor` / `prev_cursor` tokens; no total count is exposed.
 ```python
 from fastapi_toolsets.schemas import PaginatedResponse, CursorPagination
@router.get("/events")
 async def list_events() -> PaginatedResponse[EventSchema]:
    return PaginatedResponse(
        data=events,
        pagination=CursorPagination(
            next_cursor="eyJpZCI6IDQyfQ==",
            prev_cursor=None,
            items_per_page=20,
            has_more=True,
        ),
    )
 ```
 The optional `filter_attributes` field is populated when `facet_fields` are configured on the CRUD class (see [Filter attributes](crud.md#filter-attributes-facets)). It is `None` by default and can be hidden from API responses with `response_model_exclude_none=True`.
 ### [`ErrorResponse`](../reference/schemas.md#fastapi_toolsets.schemas.ErrorResponse)
@@ -1,22 +0,0 @@
 # `models`
 Here's the reference for the SQLAlchemy model mixins provided by the `models` module.
 You can import them directly from `fastapi_toolsets.models`:
 ```python
 from fastapi_toolsets.models import (
    UUIDMixin,
    CreatedAtMixin,
    UpdatedAtMixin,
    TimestampMixin,
 )
 ```
 ## ::: fastapi_toolsets.models.UUIDMixin
 ## ::: fastapi_toolsets.models.CreatedAtMixin
 ## ::: fastapi_toolsets.models.UpdatedAtMixin
 ## ::: fastapi_toolsets.models.TimestampMixin
@@ -1,4 +1,4 @@
-# `schemas`
+# `schemas` module
 Here's the reference for all response models and types provided by the `schemas` module.
@@ -12,8 +12,7 @@ from fastapi_toolsets.schemas import (
    BaseResponse,
    Response,
    ErrorResponse,
-    OffsetPagination,
+    Pagination,
    CursorPagination,
    PaginatedResponse,
 )
 ```
@@ -30,8 +29,6 @@ from fastapi_toolsets.schemas import (
 ## ::: fastapi_toolsets.schemas.ErrorResponse
-## ::: fastapi_toolsets.schemas.OffsetPagination
+## ::: fastapi_toolsets.schemas.Pagination
 ## ::: fastapi_toolsets.schemas.CursorPagination
 ## ::: fastapi_toolsets.schemas.PaginatedResponse
@@ -1,6 +1,6 @@
 [project]
 name = "fastapi-toolsets"
-version = "2.0.0"
+version = "1.3.0"
 description = "Production-ready utilities for FastAPI applications"
 readme = "README.md"
 license = "MIT"
@@ -21,4 +21,4 @@ Example usage:
        return Response(data={"user": user.username}, message="Success")
 """
-__version__ = "2.0.0"
+__version__ = "1.3.0"
@@ -1,13 +1,7 @@
 """Generic async CRUD operations for SQLAlchemy models."""
 from ..exceptions import InvalidFacetFilterError, NoSearchableFieldsError
-from ..types import (
+from ..types import FacetFieldType, JoinType, M2MFieldType, OrderByClause
    FacetFieldType,
    JoinType,
    M2MFieldType,
    OrderByClause,
    SearchFieldType,
 )
 from .factory import CrudFactory
 from .search import SearchConfig, get_searchable_fields
@@ -21,5 +15,4 @@ __all__ = [
    "NoSearchableFieldsError",
    "OrderByClause",
    "SearchConfig",
    "SearchFieldType",
 ]
@@ -10,8 +10,6 @@ from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
 from sqlalchemy.orm import DeclarativeBase
 from .exceptions import NotFoundError
 __all__ = [
    "LockMode",
    "create_db_context",
@@ -218,7 +216,7 @@ async def wait_for_row_change(
        The refreshed model instance with updated values
    Raises:
-        NotFoundError: If the row does not exist or is deleted during polling
+        LookupError: If the row does not exist or is deleted during polling
        TimeoutError: If timeout expires before a change is detected
    Example:
@@ -239,7 +237,7 @@ async def wait_for_row_change(
    """
    instance = await session.get(model, pk_value)
    if instance is None:
-        raise NotFoundError(f"{model.__name__} with pk={pk_value!r} not found")
+        raise LookupError(f"{model.__name__} with pk={pk_value!r} not found")
    if columns is not None:
        watch_cols = columns
@@ -263,7 +261,7 @@ async def wait_for_row_change(
        instance = await session.get(model, pk_value)
        if instance is None:
-            raise NotFoundError(f"{model.__name__} with pk={pk_value!r} was deleted")
+            raise LookupError(f"{model.__name__} with pk={pk_value!r} was deleted")
        current = {col: getattr(instance, col) for col in watch_cols}
        if current != initial:
@@ -6,46 +6,32 @@ from ..schemas import ApiError, ErrorResponse, ResponseStatus
 class ApiException(Exception):
-    """Base exception for API errors with structured response."""
+    """Base exception for API errors with structured response.
    Subclass this to create custom API exceptions with consistent error format.
    The exception handler will use api_error to generate the response.
    Example:
        ```python
        class CustomError(ApiException):
            api_error = ApiError(
                code=400,
                msg="Bad Request",
                desc="The request was invalid.",
                err_code="CUSTOM-400",
            )
        ```
    """
    api_error: ClassVar[ApiError]
-    def __init_subclass__(cls, abstract: bool = False, **kwargs: Any) -> None:
+    def __init__(self, detail: str | None = None):
        super().__init_subclass__(**kwargs)
        if not abstract and not hasattr(cls, "api_error"):
            raise TypeError(
                f"{cls.__name__} must define an 'api_error' class attribute. "
                "Pass abstract=True when creating intermediate base classes."
            )
    def __init__(
        self,
        detail: str | None = None,
        *,
        desc: str | None = None,
        data: Any = None,
    ) -> None:
        """Initialize the exception.
        Args:
-            detail: Optional human-readable message
+            detail: Optional override for the error message
            desc: Optional per-instance override for the ``description`` field
                in the HTTP response body.
            data: Optional per-instance override for the ``data`` field in the
                HTTP response body.
        """
-        updates: dict[str, Any] = {}
+        super().__init__(detail or self.api_error.msg)
        if detail is not None:
            updates["msg"] = detail
        if desc is not None:
            updates["desc"] = desc
        if data is not None:
            updates["data"] = data
        if updates:
            object.__setattr__(
                self, "api_error", self.__class__.api_error.model_copy(update=updates)
            )
        super().__init__(self.api_error.msg)
 class UnauthorizedError(ApiException):
@@ -106,15 +92,14 @@ class NoSearchableFieldsError(ApiException):
        """Initialize the exception.
        Args:
-            model: The model class that has no searchable fields configured.
+            model: The SQLAlchemy model class that has no searchable fields
        """
        self.model = model
-        super().__init__(
+        detail = (
            desc=(
            f"No searchable fields found for model '{model.__name__}'. "
            "Provide 'search_fields' parameter or set 'searchable_fields' on the CRUD class."
        )
-        )
+        super().__init__(detail)
 class InvalidFacetFilterError(ApiException):
@@ -131,17 +116,16 @@ class InvalidFacetFilterError(ApiException):
        """Initialize the exception.
        Args:
-            key: The unknown filter key provided by the caller.
+            key: The unknown filter key provided by the caller
-            valid_keys: Set of valid keys derived from the declared facet_fields.
+            valid_keys: Set of valid keys derived from the declared facet_fields
        """
        self.key = key
        self.valid_keys = valid_keys
-        super().__init__(
+        detail = (
            desc=(
            f"'{key}' is not a declared facet field. "
            f"Valid keys: {sorted(valid_keys) or 'none — set facet_fields on the CRUD class'}."
        )
-        )
+        super().__init__(detail)
 class InvalidOrderFieldError(ApiException):
@@ -158,14 +142,15 @@ class InvalidOrderFieldError(ApiException):
        """Initialize the exception.
        Args:
-            field: The unknown order field provided by the caller.
+            field: The unknown order field provided by the caller
-            valid_fields: List of valid field names.
+            valid_fields: List of valid field names
        """
        self.field = field
        self.valid_fields = valid_fields
-        super().__init__(
+        detail = (
-            desc=f"'{field}' is not an allowed order field. Valid fields: {valid_fields}."
+            f"'{field}' is not an allowed order field. Valid fields: {valid_fields}."
        )
        super().__init__(detail)
 def generate_error_responses(
@@ -173,39 +158,44 @@ def generate_error_responses(
 ) -> dict[int | str, dict[str, Any]]:
    """Generate OpenAPI response documentation for exceptions.
    Use this to document possible error responses for an endpoint.
    Args:
-        *errors: Exception classes that inherit from ApiException.
+        *errors: Exception classes that inherit from ApiException
    Returns:
-        Dict suitable for FastAPI's ``responses`` parameter.
+        Dict suitable for FastAPI's responses parameter
    Example:
        ```python
        from fastapi_toolsets.exceptions import generate_error_responses, UnauthorizedError, ForbiddenError
        @app.get(
            "/admin",
            responses=generate_error_responses(UnauthorizedError, ForbiddenError)
        )
        async def admin_endpoint():
            ...
        ```
    """
    responses: dict[int | str, dict[str, Any]] = {}
    for error in errors:
        api_error = error.api_error
        code = api_error.code
-        if code not in responses:
+        responses[api_error.code] = {
            responses[code] = {
            "model": ErrorResponse,
            "description": api_error.msg,
            "content": {
                "application/json": {
-                        "examples": {},
+                    "example": {
                    }
                },
            }
        responses[code]["content"]["application/json"]["examples"][
            api_error.err_code
        ] = {
            "summary": api_error.msg,
            "value": {
                        "data": api_error.data,
                        "status": ResponseStatus.FAIL.value,
                        "message": api_error.msg,
                        "description": api_error.desc,
                        "error_code": api_error.err_code,
                    }
                }
            },
        }
@@ -1,14 +1,10 @@
 """Exception handlers for FastAPI applications."""
 from collections.abc import Callable
 from typing import Any
 from fastapi import FastAPI, Request, Response, status
-from fastapi.exceptions import (
+from fastapi.exceptions import RequestValidationError, ResponseValidationError
-    HTTPException,
+from fastapi.openapi.utils import get_openapi
    RequestValidationError,
    ResponseValidationError,
 )
 from fastapi.responses import JSONResponse
 from ..schemas import ErrorResponse, ResponseStatus
@@ -22,20 +18,43 @@ _VALIDATION_LOCATION_PARAMS: frozenset[str] = frozenset(
 def init_exceptions_handlers(app: FastAPI) -> FastAPI:
    """Register exception handlers and custom OpenAPI schema on a FastAPI app.
    Installs handlers for :class:`ApiException`, validation errors, and
    unhandled exceptions, and replaces the default 422 schema with a
    consistent error format.
    Args:
-        app: FastAPI application instance.
+        app: FastAPI application instance
    Returns:
-        The same FastAPI instance (for chaining).
+        The same FastAPI instance (for chaining)
    Example:
        ```python
        from fastapi import FastAPI
        from fastapi_toolsets.exceptions import init_exceptions_handlers
        app = FastAPI()
        init_exceptions_handlers(app)
        ```
    """
    _register_exception_handlers(app)
-    _original_openapi = app.openapi
+    app.openapi = lambda: _custom_openapi(app)  # type: ignore[method-assign]
    app.openapi = lambda: _patched_openapi(app, _original_openapi)  # type: ignore[method-assign]
    return app
 def _register_exception_handlers(app: FastAPI) -> None:
-    """Register all exception handlers on a FastAPI application."""
+    """Register all exception handlers on a FastAPI application.
    Args:
        app: FastAPI application instance
    Example:
        from fastapi import FastAPI
        from fastapi_toolsets.exceptions import init_exceptions_handlers
        app = FastAPI()
        init_exceptions_handlers(app)
    """
    @app.exception_handler(ApiException)
    async def api_exception_handler(request: Request, exc: ApiException) -> Response:
@@ -47,25 +66,12 @@ def _register_exception_handlers(app: FastAPI) -> None:
            description=api_error.desc,
            error_code=api_error.err_code,
        )
        return JSONResponse(
            status_code=api_error.code,
            content=error_response.model_dump(),
        )
    @app.exception_handler(HTTPException)
    async def http_exception_handler(request: Request, exc: HTTPException) -> Response:
        """Handle Starlette/FastAPI HTTPException with a consistent error format."""
        detail = exc.detail if isinstance(exc.detail, str) else "HTTP Error"
        error_response = ErrorResponse(
            message=detail,
            error_code=f"HTTP-{exc.status_code}",
        )
        return JSONResponse(
            status_code=exc.status_code,
            content=error_response.model_dump(),
            headers=getattr(exc, "headers", None),
        )
    @app.exception_handler(RequestValidationError)
    async def request_validation_handler(
        request: Request, exc: RequestValidationError
@@ -88,6 +94,7 @@ def _register_exception_handlers(app: FastAPI) -> None:
            description="An unexpected error occurred. Please try again later.",
            error_code="SERVER-500",
        )
        return JSONResponse(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            content=error_response.model_dump(),
@@ -102,10 +109,9 @@ def _format_validation_error(
    formatted_errors = []
    for error in errors:
-        locs = error["loc"]
+        field_path = ".".join(
-        if locs and locs[0] in _VALIDATION_LOCATION_PARAMS:
+            str(loc) for loc in error["loc"] if loc not in _VALIDATION_LOCATION_PARAMS
-            locs = locs[1:]
+        )
        field_path = ".".join(str(loc) for loc in locs)
        formatted_errors.append(
            {
                "field": field_path or "root",
@@ -127,22 +133,34 @@ def _format_validation_error(
    )
-def _patched_openapi(
+def _custom_openapi(app: FastAPI) -> dict[str, Any]:
-    app: FastAPI, original_openapi: Callable[[], dict[str, Any]]
+    """Generate custom OpenAPI schema with standardized error format.
-) -> dict[str, Any]:
+
-    """Generate the OpenAPI schema and replace default 422 responses.
+    Replaces default 422 validation error responses with the custom format.
    Args:
-        app: FastAPI application instance.
+        app: FastAPI application instance
        original_openapi: The previous ``app.openapi`` callable to delegate to.
    Returns:
-        Patched OpenAPI schema dict.
+        OpenAPI schema dict
    Example:
        from fastapi import FastAPI
        from fastapi_toolsets.exceptions import init_exceptions_handlers
        app = FastAPI()
        init_exceptions_handlers(app)  # Automatically sets custom OpenAPI
    """
    if app.openapi_schema:
        return app.openapi_schema
-    openapi_schema = original_openapi()
+    openapi_schema = get_openapi(
        title=app.title,
        version=app.version,
        openapi_version=app.openapi_version,
        description=app.description,
        routes=app.routes,
    )
    for path_data in openapi_schema.get("paths", {}).values():
        for operation in path_data.values():
@@ -152,10 +170,7 @@ def _patched_openapi(
                        "description": "Validation Error",
                        "content": {
                            "application/json": {
-                                "examples": {
+                                "example": {
                                    "VAL-422": {
                                        "summary": "Validation Error",
                                        "value": {
                                    "data": {
                                        "errors": [
                                            {
@@ -169,8 +184,6 @@ def _patched_openapi(
                                    "message": "Validation Error",
                                    "description": "1 validation error(s) detected",
                                    "error_code": "VAL-422",
                                        },
                                    }
                                }
                            }
                        },
@@ -1,47 +0,0 @@
 """SQLAlchemy model mixins for common column patterns."""
 import uuid
 from datetime import datetime
 from sqlalchemy import DateTime, Uuid, func, text
 from sqlalchemy.orm import Mapped, mapped_column
 __all__ = [
    "UUIDMixin",
    "CreatedAtMixin",
    "UpdatedAtMixin",
    "TimestampMixin",
 ]
 class UUIDMixin:
    """Mixin that adds a UUID primary key auto-generated by the database."""
    id: Mapped[uuid.UUID] = mapped_column(
        Uuid,
        primary_key=True,
        server_default=text("gen_random_uuid()"),
    )
 class CreatedAtMixin:
    """Mixin that adds a ``created_at`` timestamp column."""
    created_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
    )
 class UpdatedAtMixin:
    """Mixin that adds an ``updated_at`` timestamp column."""
    updated_at: Mapped[datetime] = mapped_column(
        DateTime(timezone=True),
        server_default=func.now(),
        onupdate=func.now(),
    )
 class TimestampMixin(CreatedAtMixin, UpdatedAtMixin):
    """Mixin that combines ``created_at`` and ``updated_at`` timestamp columns."""
@@ -430,7 +430,7 @@ class TestNoSearchableFieldsError:
        from fastapi_toolsets.exceptions import NoSearchableFieldsError
        error = NoSearchableFieldsError(User)
-        assert "User" in error.api_error.desc
+        assert "User" in str(error)
        assert error.model is User
    def test_error_raised_when_no_fields(self):
@@ -454,7 +454,7 @@ class TestNoSearchableFieldsError:
            build_search_filters(NoStringModel, "test")
        assert exc_info.value.model is NoStringModel
-        assert "NoStringModel" in exc_info.value.api_error.desc
+        assert "NoStringModel" in str(exc_info.value)
 class TestGetSearchableFields:
@@ -14,7 +14,6 @@ from fastapi_toolsets.db import (
    lock_tables,
    wait_for_row_change,
 )
 from fastapi_toolsets.exceptions import NotFoundError
 from .conftest import DATABASE_URL, Base, Role, RoleCrud, User
@@ -308,9 +307,9 @@ class TestWaitForRowChange:
    @pytest.mark.anyio
    async def test_nonexistent_row_raises(self, db_session: AsyncSession):
-        """Raises NotFoundError when the row does not exist."""
+        """Raises LookupError when the row does not exist."""
        fake_id = uuid.uuid4()
-        with pytest.raises(NotFoundError, match="not found"):
+        with pytest.raises(LookupError, match="not found"):
            await wait_for_row_change(db_session, Role, fake_id, interval=0.05)
    @pytest.mark.anyio
@@ -327,7 +326,7 @@ class TestWaitForRowChange:
    @pytest.mark.anyio
    async def test_deleted_row_raises(self, db_session: AsyncSession, engine):
-        """Raises NotFoundError when the row is deleted during polling."""
+        """Raises LookupError when the row is deleted during polling."""
        role = Role(name="delete_role")
        db_session.add(role)
        await db_session.commit()
@@ -341,6 +340,6 @@ class TestWaitForRowChange:
                await other.commit()
        delete_task = asyncio.create_task(delete_later())
-        with pytest.raises(NotFoundError):
+        with pytest.raises(LookupError):
            await wait_for_row_change(db_session, Role, role.id, interval=0.05)
        await delete_task
@@ -2,7 +2,6 @@
 import pytest
 from fastapi import FastAPI
 from fastapi.exceptions import HTTPException
 from fastapi.testclient import TestClient
 from fastapi_toolsets.exceptions import (
@@ -37,8 +36,8 @@ class TestApiException:
        assert error.api_error.msg == "I'm a teapot"
        assert str(error) == "I'm a teapot"
-    def test_detail_overrides_msg_and_str(self):
+    def test_custom_detail_message(self):
-        """detail sets both str(exc) and api_error.msg; class-level msg is unchanged."""
+        """Custom detail overrides default message."""
        class CustomError(ApiException):
            api_error = ApiError(
@@ -48,172 +47,8 @@ class TestApiException:
                err_code="BAD-400",
            )
-        error = CustomError("Widget not found")
+        error = CustomError("Custom message")
-        assert str(error) == "Widget not found"
+        assert str(error) == "Custom message"
        assert error.api_error.msg == "Widget not found"
        assert CustomError.api_error.msg == "Bad Request"  # class unchanged
    def test_desc_override(self):
        """desc kwarg overrides api_error.desc on the instance only."""
        class MyError(ApiException):
            api_error = ApiError(
                code=400, msg="Error", desc="Default.", err_code="ERR-400"
            )
        err = MyError(desc="Custom desc.")
        assert err.api_error.desc == "Custom desc."
        assert MyError.api_error.desc == "Default."  # class unchanged
    def test_data_override(self):
        """data kwarg sets api_error.data on the instance only."""
        class MyError(ApiException):
            api_error = ApiError(
                code=400, msg="Error", desc="Default.", err_code="ERR-400"
            )
        err = MyError(data={"key": "value"})
        assert err.api_error.data == {"key": "value"}
        assert MyError.api_error.data is None  # class unchanged
    def test_desc_and_data_override(self):
        """detail, desc and data can all be overridden together."""
        class MyError(ApiException):
            api_error = ApiError(
                code=400, msg="Error", desc="Default.", err_code="ERR-400"
            )
        err = MyError("custom msg", desc="New desc.", data={"x": 1})
        assert str(err) == "custom msg"
        assert err.api_error.msg == "custom msg"  # detail also updates msg
        assert err.api_error.desc == "New desc."
        assert err.api_error.data == {"x": 1}
        assert err.api_error.code == 400  # other fields unchanged
    def test_class_api_error_not_mutated_after_instance_override(self):
        """Raising with desc/data does not mutate the class-level api_error."""
        class MyError(ApiException):
            api_error = ApiError(
                code=400, msg="Error", desc="Default.", err_code="ERR-400"
            )
        MyError(desc="Changed", data={"x": 1})
        assert MyError.api_error.desc == "Default."
        assert MyError.api_error.data is None
    def test_subclass_uses_super_with_desc_and_data(self):
        """Subclasses can delegate detail/desc/data to super().__init__()."""
        class BuildValidationError(ApiException):
            api_error = ApiError(
                code=422,
                msg="Build Validation Error",
                desc="The build configuration is invalid.",
                err_code="BUILD-422",
            )
            def __init__(self, *errors: str) -> None:
                super().__init__(
                    f"{len(errors)} validation error(s)",
                    desc=", ".join(errors),
                    data={"errors": [{"message": e} for e in errors]},
                )
        err = BuildValidationError("Field A is required", "Field B is invalid")
        assert str(err) == "2 validation error(s)"
        assert err.api_error.msg == "2 validation error(s)"  # detail set msg
        assert err.api_error.desc == "Field A is required, Field B is invalid"
        assert err.api_error.data == {
            "errors": [
                {"message": "Field A is required"},
                {"message": "Field B is invalid"},
            ]
        }
        assert err.api_error.code == 422  # other fields unchanged
    def test_detail_desc_data_in_http_response(self):
        """detail/desc/data overrides all appear correctly in the FastAPI HTTP response."""
        class DynamicError(ApiException):
            api_error = ApiError(
                code=400, msg="Error", desc="Default.", err_code="ERR-400"
            )
            def __init__(self, message: str) -> None:
                super().__init__(
                    message,
                    desc=f"Detail: {message}",
                    data={"reason": message},
                )
        app = FastAPI()
        init_exceptions_handlers(app)
        @app.get("/error")
        async def raise_error():
            raise DynamicError("something went wrong")
        client = TestClient(app)
        response = client.get("/error")
        assert response.status_code == 400
        body = response.json()
        assert body["message"] == "something went wrong"
        assert body["description"] == "Detail: something went wrong"
        assert body["data"] == {"reason": "something went wrong"}
 class TestApiExceptionGuard:
    """Tests for the __init_subclass__ api_error guard."""
    def test_missing_api_error_raises_type_error(self):
        """Defining a subclass without api_error raises TypeError at class creation time."""
        with pytest.raises(
            TypeError, match="must define an 'api_error' class attribute"
        ):
            class BrokenError(ApiException):
                pass
    def test_abstract_subclass_skips_guard(self):
        """abstract=True allows intermediate base classes without api_error."""
        class BaseGroupError(ApiException, abstract=True):
            pass
        # Concrete child must still define it
        class ConcreteError(BaseGroupError):
            api_error = ApiError(
                code=400, msg="Error", desc="Desc.", err_code="ERR-400"
            )
        err = ConcreteError()
        assert err.api_error.code == 400
    def test_abstract_child_still_requires_api_error_on_concrete(self):
        """Concrete subclass of an abstract class must define api_error."""
        class Base(ApiException, abstract=True):
            pass
        with pytest.raises(
            TypeError, match="must define an 'api_error' class attribute"
        ):
            class Concrete(Base):
                pass
    def test_inherited_api_error_satisfies_guard(self):
        """Subclass that inherits api_error from a parent does not need its own."""
        class ConcreteError(NotFoundError):
            pass
        err = ConcreteError()
        assert err.api_error.code == 404
 class TestBuiltInExceptions:
@@ -255,7 +90,7 @@ class TestGenerateErrorResponses:
        assert responses[404]["description"] == "Not Found"
    def test_generates_multiple_responses(self):
-        """Generates responses for multiple exceptions with distinct status codes."""
+        """Generates responses for multiple exceptions."""
        responses = generate_error_responses(
            UnauthorizedError,
            ForbiddenError,
@@ -266,24 +101,15 @@ class TestGenerateErrorResponses:
        assert 403 in responses
        assert 404 in responses
-    def test_response_has_named_example(self):
+    def test_response_has_example(self):
-        """Generated response uses named examples keyed by err_code."""
+        """Generated response includes example."""
        responses = generate_error_responses(NotFoundError)
-        examples = responses[404]["content"]["application/json"]["examples"]
+        example = responses[404]["content"]["application/json"]["example"]
-        assert "RES-404" in examples
+        assert example["status"] == "FAIL"
-        value = examples["RES-404"]["value"]
+        assert example["error_code"] == "RES-404"
-        assert value["status"] == "FAIL"
+        assert example["message"] == "Not Found"
-        assert value["error_code"] == "RES-404"
+        assert example["data"] is None
        assert value["message"] == "Not Found"
        assert value["data"] is None
    def test_response_example_has_summary(self):
        """Each named example carries a summary equal to api_error.msg."""
        responses = generate_error_responses(NotFoundError)
        example = responses[404]["content"]["application/json"]["examples"]["RES-404"]
        assert example["summary"] == "Not Found"
    def test_response_example_with_data(self):
        """Generated response includes data when set on ApiError."""
@@ -298,49 +124,9 @@ class TestGenerateErrorResponses:
            )
        responses = generate_error_responses(ErrorWithData)
-        value = responses[400]["content"]["application/json"]["examples"]["BAD-400"][
+        example = responses[400]["content"]["application/json"]["example"]
            "value"
        ]
-        assert value["data"] == {"details": "some context"}
+        assert example["data"] == {"details": "some context"}
    def test_two_errors_same_code_both_present(self):
        """Two exceptions with the same HTTP code produce two named examples."""
        class BadRequestA(ApiException):
            api_error = ApiError(
                code=400, msg="Bad A", desc="Reason A.", err_code="ERR-A"
            )
        class BadRequestB(ApiException):
            api_error = ApiError(
                code=400, msg="Bad B", desc="Reason B.", err_code="ERR-B"
            )
        responses = generate_error_responses(BadRequestA, BadRequestB)
        assert 400 in responses
        examples = responses[400]["content"]["application/json"]["examples"]
        assert "ERR-A" in examples
        assert "ERR-B" in examples
        assert examples["ERR-A"]["value"]["message"] == "Bad A"
        assert examples["ERR-B"]["value"]["message"] == "Bad B"
    def test_two_errors_same_code_single_top_level_entry(self):
        """Two exceptions with the same HTTP code produce exactly one top-level entry."""
        class BadRequestA(ApiException):
            api_error = ApiError(
                code=400, msg="Bad A", desc="Reason A.", err_code="ERR-A"
            )
        class BadRequestB(ApiException):
            api_error = ApiError(
                code=400, msg="Bad B", desc="Reason B.", err_code="ERR-B"
            )
        responses = generate_error_responses(BadRequestA, BadRequestB)
        assert len([k for k in responses if k == 400]) == 1
 class TestInitExceptionsHandlers:
@@ -464,68 +250,13 @@ class TestInitExceptionsHandlers:
        assert data["status"] == "FAIL"
        assert data["error_code"] == "SERVER-500"
    def test_handles_http_exception(self):
        """Handles starlette HTTPException with consistent ErrorResponse envelope."""
        app = FastAPI()
        init_exceptions_handlers(app)
        @app.get("/protected")
        async def protected():
            raise HTTPException(status_code=403, detail="Forbidden")
        client = TestClient(app)
        response = client.get("/protected")
        assert response.status_code == 403
        data = response.json()
        assert data["status"] == "FAIL"
        assert data["error_code"] == "HTTP-403"
        assert data["message"] == "Forbidden"
    def test_handles_http_exception_404_from_route(self):
        """HTTPException(404) raised inside a route uses the consistent ErrorResponse envelope."""
        app = FastAPI()
        init_exceptions_handlers(app)
        @app.get("/items/{item_id}")
        async def get_item(item_id: int):
            raise HTTPException(status_code=404, detail="Item not found")
        client = TestClient(app)
        response = client.get("/items/99")
        assert response.status_code == 404
        data = response.json()
        assert data["status"] == "FAIL"
        assert data["error_code"] == "HTTP-404"
        assert data["message"] == "Item not found"
    def test_handles_http_exception_forwards_headers(self):
        """HTTPException with WWW-Authenticate header forwards it in the response."""
        app = FastAPI()
        init_exceptions_handlers(app)
        @app.get("/secure")
        async def secure():
            raise HTTPException(
                status_code=401,
                detail="Not authenticated",
                headers={"WWW-Authenticate": "Bearer"},
            )
        client = TestClient(app)
        response = client.get("/secure")
        assert response.status_code == 401
        assert response.headers.get("www-authenticate") == "Bearer"
    def test_custom_openapi_schema(self):
-        """Customises OpenAPI schema for 422 responses using named examples."""
+        """Customizes OpenAPI schema for 422 responses."""
        from pydantic import BaseModel
        app = FastAPI()
        init_exceptions_handlers(app)
        from pydantic import BaseModel
        class Item(BaseModel):
            name: str
@@ -538,128 +269,8 @@ class TestInitExceptionsHandlers:
        post_op = openapi["paths"]["/items"]["post"]
        assert "422" in post_op["responses"]
        resp_422 = post_op["responses"]["422"]
-        examples = resp_422["content"]["application/json"]["examples"]
+        example = resp_422["content"]["application/json"]["example"]
-        assert "VAL-422" in examples
+        assert example["error_code"] == "VAL-422"
        assert examples["VAL-422"]["value"]["error_code"] == "VAL-422"
    def test_custom_openapi_preserves_app_metadata(self):
        """_patched_openapi preserves custom FastAPI app-level metadata."""
        app = FastAPI(
            title="My API",
            version="2.0.0",
            description="Custom description",
        )
        init_exceptions_handlers(app)
        schema = app.openapi()
        assert schema["info"]["title"] == "My API"
        assert schema["info"]["version"] == "2.0.0"
    def test_handles_response_validation_error(self):
        """Handles ResponseValidationError with a structured 422 response."""
        from pydantic import BaseModel
        class CountResponse(BaseModel):
            count: int
        app = FastAPI()
        init_exceptions_handlers(app)
        @app.get("/broken", response_model=CountResponse)
        async def broken():
            return {"count": "not-a-number"}  # triggers ResponseValidationError
        client = TestClient(app, raise_server_exceptions=False)
        response = client.get("/broken")
        assert response.status_code == 422
        data = response.json()
        assert data["status"] == "FAIL"
        assert data["error_code"] == "VAL-422"
        assert "errors" in data["data"]
    def test_handles_validation_error_with_non_standard_loc(self):
        """Validation error with empty loc tuple maps the field to 'root'."""
        from fastapi.exceptions import RequestValidationError
        app = FastAPI()
        init_exceptions_handlers(app)
        @app.get("/root-error")
        async def root_error():
            raise RequestValidationError(
                [
                    {
                        "type": "custom",
                        "loc": (),
                        "msg": "root level error",
                        "input": None,
                        "url": "",
                    }
                ]
            )
        client = TestClient(app)
        response = client.get("/root-error")
        assert response.status_code == 422
        data = response.json()
        assert data["data"]["errors"][0]["field"] == "root"
    def test_openapi_schema_cached_after_first_call(self):
        """app.openapi() returns the cached schema on subsequent calls."""
        from pydantic import BaseModel
        app = FastAPI()
        init_exceptions_handlers(app)
        class Item(BaseModel):
            name: str
        @app.post("/items")
        async def create_item(item: Item):
            return item
        schema_first = app.openapi()
        schema_second = app.openapi()
        assert schema_first is schema_second
    def test_openapi_skips_operations_without_422(self):
        """_patched_openapi leaves operations that have no 422 response unchanged."""
        app = FastAPI()
        init_exceptions_handlers(app)
        @app.get("/ping")
        async def ping():
            return {"ok": True}
        schema = app.openapi()
        get_op = schema["paths"]["/ping"]["get"]
        assert "422" not in get_op["responses"]
        assert "200" in get_op["responses"]
    def test_openapi_skips_non_dict_path_item_values(self):
        """_patched_openapi ignores non-dict values in path items (e.g. path-level parameters)."""
        from fastapi_toolsets.exceptions.handler import _patched_openapi
        app = FastAPI()
        def fake_openapi() -> dict:
            return {
                "paths": {
                    "/items": {
                        "parameters": [
                            {"name": "q", "in": "query"}
                        ],  # list, not a dict
                        "get": {"responses": {"200": {"description": "OK"}}},
                    }
                }
            }
        schema = _patched_openapi(app, fake_openapi)
        # The list value was skipped without error; the GET operation is intact
        assert schema["paths"]["/items"]["parameters"] == [{"name": "q", "in": "query"}]
        assert "422" not in schema["paths"]["/items"]["get"]["responses"]
 class TestExceptionIntegration:
@@ -741,12 +352,12 @@ class TestInvalidOrderFieldError:
        assert error.field == "unknown"
        assert error.valid_fields == ["name", "created_at"]
-    def test_description_contains_field_and_valid_fields(self):
+    def test_message_contains_field_and_valid_fields(self):
-        """api_error.desc mentions the bad field and valid options."""
+        """Exception message mentions the bad field and valid options."""
        error = InvalidOrderFieldError("bad_field", ["name", "email"])
-        assert "bad_field" in error.api_error.desc
+        assert "bad_field" in str(error)
-        assert "name" in error.api_error.desc
+        assert "name" in str(error)
-        assert "email" in error.api_error.desc
+        assert "email" in str(error)
    def test_handled_as_422_by_exception_handler(self):
        """init_exceptions_handlers turns InvalidOrderFieldError into a 422 response."""
@@ -1,247 +0,0 @@
 """Tests for fastapi_toolsets.models mixins."""
 import uuid
 import pytest
 from sqlalchemy import String
 from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
 from fastapi_toolsets.models import (
    CreatedAtMixin,
    TimestampMixin,
    UUIDMixin,
    UpdatedAtMixin,
 )
 from .conftest import DATABASE_URL
 class MixinBase(DeclarativeBase):
    pass
 class UUIDModel(MixinBase, UUIDMixin):
    __tablename__ = "mixin_uuid_models"
    name: Mapped[str] = mapped_column(String(50))
 class UpdatedAtModel(MixinBase, UpdatedAtMixin):
    __tablename__ = "mixin_updated_at_models"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50))
 class CreatedAtModel(MixinBase, CreatedAtMixin):
    __tablename__ = "mixin_created_at_models"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50))
 class TimestampModel(MixinBase, TimestampMixin):
    __tablename__ = "mixin_timestamp_models"
    id: Mapped[int] = mapped_column(primary_key=True, autoincrement=True)
    name: Mapped[str] = mapped_column(String(50))
 class FullMixinModel(MixinBase, UUIDMixin, UpdatedAtMixin):
    __tablename__ = "mixin_full_models"
    name: Mapped[str] = mapped_column(String(50))
@pytest.fixture(scope="function")
 async def mixin_session():
    engine = create_async_engine(DATABASE_URL, echo=False)
    async with engine.begin() as conn:
        await conn.run_sync(MixinBase.metadata.create_all)
    session_factory = async_sessionmaker(engine, expire_on_commit=False)
    session = session_factory()
    try:
        yield session
    finally:
        await session.close()
        async with engine.begin() as conn:
            await conn.run_sync(MixinBase.metadata.drop_all)
        await engine.dispose()
 class TestUUIDMixin:
    @pytest.mark.anyio
    async def test_uuid_generated_by_db(self, mixin_session):
        """UUID is generated server-side and populated after flush."""
        obj = UUIDModel(name="test")
        mixin_session.add(obj)
        await mixin_session.flush()
        assert obj.id is not None
        assert isinstance(obj.id, uuid.UUID)
    @pytest.mark.anyio
    async def test_uuid_is_primary_key(self):
        """UUIDMixin adds id as primary key column."""
        pk_cols = [c.name for c in UUIDModel.__table__.primary_key]
        assert pk_cols == ["id"]
    @pytest.mark.anyio
    async def test_each_row_gets_unique_uuid(self, mixin_session):
        """Each inserted row gets a distinct UUID."""
        a = UUIDModel(name="a")
        b = UUIDModel(name="b")
        mixin_session.add_all([a, b])
        await mixin_session.flush()
        assert a.id != b.id
    @pytest.mark.anyio
    async def test_uuid_server_default_set(self):
        """Column has gen_random_uuid() as server default."""
        col = UUIDModel.__table__.c["id"]
        assert col.server_default is not None
        assert "gen_random_uuid" in str(col.server_default.arg)
 class TestUpdatedAtMixin:
    @pytest.mark.anyio
    async def test_updated_at_set_on_insert(self, mixin_session):
        """updated_at is populated after insert."""
        obj = UpdatedAtModel(name="initial")
        mixin_session.add(obj)
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        assert obj.updated_at is not None
        assert obj.updated_at.tzinfo is not None  # timezone-aware
    @pytest.mark.anyio
    async def test_updated_at_changes_on_update(self, mixin_session):
        """updated_at is updated when the row is modified."""
        obj = UpdatedAtModel(name="initial")
        mixin_session.add(obj)
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        original_ts = obj.updated_at
        obj.name = "modified"
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        assert obj.updated_at >= original_ts
    @pytest.mark.anyio
    async def test_updated_at_column_is_not_nullable(self):
        """updated_at column is non-nullable."""
        col = UpdatedAtModel.__table__.c["updated_at"]
        assert not col.nullable
    @pytest.mark.anyio
    async def test_updated_at_has_server_default(self):
        """updated_at column has a server-side default."""
        col = UpdatedAtModel.__table__.c["updated_at"]
        assert col.server_default is not None
    @pytest.mark.anyio
    async def test_updated_at_has_onupdate(self):
        """updated_at column has an onupdate clause."""
        col = UpdatedAtModel.__table__.c["updated_at"]
        assert col.onupdate is not None
 class TestCreatedAtMixin:
    @pytest.mark.anyio
    async def test_created_at_set_on_insert(self, mixin_session):
        """created_at is populated after insert."""
        obj = CreatedAtModel(name="new")
        mixin_session.add(obj)
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        assert obj.created_at is not None
        assert obj.created_at.tzinfo is not None  # timezone-aware
    @pytest.mark.anyio
    async def test_created_at_not_changed_on_update(self, mixin_session):
        """created_at is not modified when the row is updated."""
        obj = CreatedAtModel(name="original")
        mixin_session.add(obj)
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        original_ts = obj.created_at
        obj.name = "updated"
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        assert obj.created_at == original_ts
    @pytest.mark.anyio
    async def test_created_at_column_is_not_nullable(self):
        """created_at column is non-nullable."""
        col = CreatedAtModel.__table__.c["created_at"]
        assert not col.nullable
    @pytest.mark.anyio
    async def test_created_at_has_no_onupdate(self):
        """created_at column has no onupdate clause."""
        col = CreatedAtModel.__table__.c["created_at"]
        assert col.onupdate is None
 class TestTimestampMixin:
    @pytest.mark.anyio
    async def test_both_columns_set_on_insert(self, mixin_session):
        """created_at and updated_at are both populated after insert."""
        obj = TimestampModel(name="new")
        mixin_session.add(obj)
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        assert obj.created_at is not None
        assert obj.updated_at is not None
    @pytest.mark.anyio
    async def test_created_at_stable_updated_at_changes_on_update(self, mixin_session):
        """On update: created_at stays the same, updated_at advances."""
        obj = TimestampModel(name="original")
        mixin_session.add(obj)
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        original_created = obj.created_at
        original_updated = obj.updated_at
        obj.name = "modified"
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        assert obj.created_at == original_created
        assert obj.updated_at >= original_updated
    @pytest.mark.anyio
    async def test_timestamp_mixin_has_both_columns(self):
        """TimestampModel exposes both created_at and updated_at columns."""
        col_names = {c.name for c in TimestampModel.__table__.columns}
        assert "created_at" in col_names
        assert "updated_at" in col_names
 class TestFullMixinModel:
    @pytest.mark.anyio
    async def test_combined_mixins_work_together(self, mixin_session):
        """UUIDMixin and UpdatedAtMixin can be combined on the same model."""
        obj = FullMixinModel(name="combined")
        mixin_session.add(obj)
        await mixin_session.flush()
        await mixin_session.refresh(obj)
        assert isinstance(obj.id, uuid.UUID)
        assert obj.updated_at is not None
        assert obj.updated_at.tzinfo is not None
@@ -251,7 +251,7 @@ wheels = [
 [[package]]
 name = "fastapi-toolsets"
-version = "2.0.0"
+version = "1.3.0"
 source = { editable = "." }
 dependencies = [
    { name = "asyncpg" },
@@ -77,10 +77,6 @@ md_in_html = {}
 "pymdownx.tasklist" = {custom_checkbox = true}
 "pymdownx.tilde" = {}
 [project.markdown_extensions.pymdownx.emoji]
 emoji_index = "zensical.extensions.emoji.twemoji"
 emoji_generator = "zensical.extensions.emoji.to_svg"
 [project.markdown_extensions."pymdownx.highlight"]
 anchor_linenums = true
 line_spans = "__span"
@@ -99,49 +95,3 @@ permalink = true
 [project.markdown_extensions."pymdownx.snippets"]
 base_path = ["."]
 check_paths = true
 [[project.nav]]
 Home = "index.md"
 [[project.nav]]
 Modules = [
    {CLI = "module/cli.md"},
    {CRUD = "module/crud.md"},
    {Database = "module/db.md"},
    {Dependencies = "module/dependencies.md"},
    {Exceptions = "module/exceptions.md"},
    {Fixtures = "module/fixtures.md"},
    {Logger = "module/logger.md"},
    {Metrics = "module/metrics.md"},
    {Models = "module/models.md"},
    {Pytest = "module/pytest.md"},
    {Schemas = "module/schemas.md"},
 ]
 [[project.nav]]
 Reference = [
    {CLI = "reference/cli.md"},
    {CRUD = "reference/crud.md"},
    {Database = "reference/db.md"},
    {Dependencies = "reference/dependencies.md"},
    {Exceptions = "reference/exceptions.md"},
    {Fixtures = "reference/fixtures.md"},
    {Logger = "reference/logger.md"},
    {Metrics = "reference/metrics.md"},
    {Models = "reference/models.md"},
    {Pytest = "reference/pytest.md"},
    {Schemas = "reference/schemas.md"},
 ]
 [[project.nav]]
 Examples = [
    {"Pagination & Search" = "examples/pagination-search.md"},
 ]
 [[project.nav]]
 Migration = [
    {"v2.0" = "migration/v2.md"},
 ]
 [[project.nav]]
 "Changelog ↗" = "https://github.com/d3vyce/fastapi-toolsets/releases"
Author	SHA1	Message	Date
d3vyce	0fc86d3c34	refactor: simplify and deduplicate across crud, metrics, cli, and exceptions	2026-03-01 09:47:36 -05:00
d3vyce	82ef96082e	test: add missing tests for fixtures/utils.py	2026-03-01 08:05:20 -05:00
d3vyce	e0828c7e71	refactor: centralize type aliases in types.py and simplify crud layer	2026-03-01 07:46:49 -05:00
d3vyce	59d028d00e	refactor: remove deprecated parameter and function	2026-03-01 07:21:07 -05:00