Version 2.3.0

README.md

@@ -48,8 +48,8 @@ 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`)
+- **Model Mixins**: SQLAlchemy mixins for common column patterns (`UUIDMixin`, `UUIDv7Mixin`, `CreatedAtMixin`, `UpdatedAtMixin`, `TimestampMixin`)
-- **Standardized API Responses**: Consistent response format with `Response`, `PaginatedResponse`, and `PydanticBase`
+- **Standardized API Responses**: Consistent response format with `Response`, `ErrorResponse`, `PaginatedResponse`, `CursorPaginatedResponse` and `OffsetPaginatedResponse`.
 - **Exception Handling**: Structured error responses with automatic OpenAPI documentation
 - **Logging**: Logging configuration with uvicorn integration via `configure_logging` and `get_logger`
 

docs/index.md

@@ -48,8 +48,8 @@ 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`)
+- **Model Mixins**: SQLAlchemy mixins for common column patterns (`UUIDMixin`, `UUIDv7Mixin`, `CreatedAtMixin`, `UpdatedAtMixin`, `TimestampMixin`)
-- **Standardized API Responses**: Consistent response format with `Response`, `PaginatedResponse`, and `PydanticBase`
+- **Standardized API Responses**: Consistent response format with `Response`, `ErrorResponse`, `PaginatedResponse`, `CursorPaginatedResponse` and `OffsetPaginatedResponse`.
 - **Exception Handling**: Structured error responses with automatic OpenAPI documentation
 - **Logging**: Logging configuration with uvicorn integration via `configure_logging` and `get_logger`
 

docs/module/schemas.md

@@ -102,7 +102,9 @@ async def list_events(
 
 #### [`PaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse)
 
-Base class and return type for endpoints that support **both** pagination strategies via a `pagination_type` query parameter (using [`paginate()`](crud.md#unified-paginate--both-strategies-on-one-endpoint))
+Return type for endpoints that support **both** pagination strategies via a `pagination_type` query parameter (using [`paginate()`](crud.md#unified-paginate--both-strategies-on-one-endpoint)).
+
+When used as a return annotation, `PaginatedResponse[T]` automatically expands to `Annotated[Union[CursorPaginatedResponse[T], OffsetPaginatedResponse[T]], Field(discriminator="pagination_type")]`, so FastAPI emits a proper `oneOf` + discriminator in the OpenAPI schema with no extra boilerplate:
 
 ```python
 from fastapi_toolsets.crud import PaginationType

pyproject.toml

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

src/fastapi_toolsets/__init__.py

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

src/fastapi_toolsets/crud/factory.py

@@ -9,6 +9,7 @@ import uuid as uuid_module
 from collections.abc import Awaitable, Callable, Sequence
 from datetime import date, datetime
 from decimal import Decimal
+from enum import Enum
 from typing import Any, ClassVar, Generic, Literal, Self, cast, overload
 
 from fastapi import Query
@@ -49,17 +50,43 @@ from .search import (
 )
 
 
-def _encode_cursor(value: Any, *, direction: str = "next") -> str:
+class _CursorDirection(str, Enum):
+    NEXT = "next"
+    PREV = "prev"
+
+
+def _encode_cursor(
+    value: Any, *, direction: _CursorDirection = _CursorDirection.NEXT
+) -> str:
     """Encode a cursor column value and navigation direction as a base64 string."""
     return base64.b64encode(
         json.dumps({"val": str(value), "dir": direction}).encode()
     ).decode()
 
 
-def _decode_cursor(cursor: str) -> tuple[str, str]:
+def _decode_cursor(cursor: str) -> tuple[str, _CursorDirection]:
     """Decode a cursor base64 string into ``(raw_value, direction)``."""
     payload = json.loads(base64.b64decode(cursor.encode()).decode())
-    return payload["val"], payload["dir"]
+    return payload["val"], _CursorDirection(payload["dir"])
+
+
+def _parse_cursor_value(raw_val: str, col_type: Any) -> Any:
+    """Parse a raw cursor string value back into the appropriate Python type."""
+    if isinstance(col_type, Integer):
+        return int(raw_val)
+    if isinstance(col_type, Uuid):
+        return uuid_module.UUID(raw_val)
+    if isinstance(col_type, DateTime):
+        return datetime.fromisoformat(raw_val)
+    if isinstance(col_type, Date):
+        return date.fromisoformat(raw_val)
+    if isinstance(col_type, (Float, Numeric)):
+        return Decimal(raw_val)
+    raise ValueError(
+        f"Unsupported cursor column type: {type(col_type).__name__!r}. "
+        "Supported types: Integer, BigInteger, SmallInteger, Uuid, "
+        "DateTime, Date, Float, Numeric."
+    )
 
 
 def _apply_joins(q: Any, joins: JoinType | None, outer_join: bool) -> Any:
@@ -108,8 +135,9 @@ class AsyncCrud(Generic[ModelType]):
         if raw_fields is None:
             cls.searchable_fields = [pk_col]
         else:
-            existing_keys = {f.key for f in raw_fields if not isinstance(f, tuple)}
+            if not any(
-            if pk_key not in existing_keys:
+                not isinstance(f, tuple) and f.key == pk_key for f in raw_fields
+            ):
                 cls.searchable_fields = [pk_col, *raw_fields]
 
     @classmethod
@@ -1048,27 +1076,12 @@ class AsyncCrud(Generic[ModelType]):
         cursor_column: Any = cls.cursor_column
         cursor_col_name: str = cursor_column.key
 
-        direction = "next"
+        direction = _CursorDirection.NEXT
         if cursor is not None:
             raw_val, direction = _decode_cursor(cursor)
             col_type = cursor_column.property.columns[0].type
-            if isinstance(col_type, Integer):
+            cursor_val: Any = _parse_cursor_value(raw_val, col_type)
-                cursor_val: Any = int(raw_val)
+            if direction is _CursorDirection.PREV:
-            elif isinstance(col_type, Uuid):
-                cursor_val = uuid_module.UUID(raw_val)
-            elif isinstance(col_type, DateTime):
-                cursor_val = datetime.fromisoformat(raw_val)
-            elif isinstance(col_type, Date):
-                cursor_val = date.fromisoformat(raw_val)
-            elif isinstance(col_type, (Float, Numeric)):
-                cursor_val = Decimal(raw_val)
-            else:
-                raise ValueError(
-                    f"Unsupported cursor column type: {type(col_type).__name__!r}. "
-                    "Supported types: Integer, BigInteger, SmallInteger, Uuid, "
-                    "DateTime, Date, Float, Numeric."
-                )
-            if direction == "prev":
                 filters.append(cursor_column < cursor_val)
             else:
                 filters.append(cursor_column > cursor_val)
@@ -1099,7 +1112,7 @@ class AsyncCrud(Generic[ModelType]):
             q = q.options(*resolved)
 
         # Cursor column is always the primary sort; reverse direction for prev traversal
-        if direction == "prev":
+        if direction is _CursorDirection.PREV:
             q = q.order_by(cursor_column.desc())
         else:
             q = q.order_by(cursor_column)
@@ -1115,32 +1128,34 @@ class AsyncCrud(Generic[ModelType]):
         items_page = raw_items[:items_per_page]
 
         # Restore ascending order when traversing backward
-        if direction == "prev":
+        if direction is _CursorDirection.PREV:
             items_page = list(reversed(items_page))
 
         # next_cursor: points past the last item in ascending order
         next_cursor: str | None = None
-        if direction == "next":
+        if direction is _CursorDirection.NEXT:
             if has_more and items_page:
                 next_cursor = _encode_cursor(
-                    getattr(items_page[-1], cursor_col_name), direction="next"
+                    getattr(items_page[-1], cursor_col_name),
+                    direction=_CursorDirection.NEXT,
                 )
         else:
             # Going backward: always provide a next_cursor to allow returning forward
             if items_page:
                 next_cursor = _encode_cursor(
-                    getattr(items_page[-1], cursor_col_name), direction="next"
+                    getattr(items_page[-1], cursor_col_name),
+                    direction=_CursorDirection.NEXT,
                 )
 
         # prev_cursor: points before the first item in ascending order
         prev_cursor: str | None = None
-        if direction == "next" and cursor is not None and items_page:
+        if direction is _CursorDirection.NEXT and cursor is not None and items_page:
             prev_cursor = _encode_cursor(
-                getattr(items_page[0], cursor_col_name), direction="prev"
+                getattr(items_page[0], cursor_col_name), direction=_CursorDirection.PREV
             )
-        elif direction == "prev" and has_more and items_page:
+        elif direction is _CursorDirection.PREV and has_more and items_page:
             prev_cursor = _encode_cursor(
-                getattr(items_page[0], cursor_col_name), direction="prev"
+                getattr(items_page[0], cursor_col_name), direction=_CursorDirection.PREV
             )
 
         items: list[Any] = [schema.model_validate(item) for item in items_page]

src/fastapi_toolsets/schemas.py

@@ -1,9 +1,9 @@
 """Base Pydantic schemas for API responses."""
 
 from enum import Enum
-from typing import Any, ClassVar, Generic, Literal
+from typing import Annotated, Any, ClassVar, Generic, Literal, TypeVar, Union
 
-from pydantic import BaseModel, ConfigDict
+from pydantic import BaseModel, ConfigDict, Field
 
 from .types import DataT
 
@@ -138,9 +138,11 @@ class PaginatedResponse(BaseResponse, Generic[DataT]):
 
     Base class and return type for endpoints that support both pagination
     strategies.  Use :class:`OffsetPaginatedResponse` or
-    :class:`CursorPaginatedResponse` when the strategy is fixed;  use
+    :class:`CursorPaginatedResponse` when the strategy is fixed.
-    ``PaginatedResponse`` as the return annotation for unified endpoints that
+
-    dispatch via :meth:`~fastapi_toolsets.crud.factory.AsyncCrud.paginate`.
+    When used as ``PaginatedResponse[T]`` in a return annotation, subscripting
+    returns ``Annotated[Union[CursorPaginatedResponse[T], OffsetPaginatedResponse[T]], Field(discriminator="pagination_type")]``
+    so FastAPI emits a proper ``oneOf`` + discriminator in the OpenAPI schema.
     """
 
     data: list[DataT]
@@ -148,6 +150,22 @@ class PaginatedResponse(BaseResponse, Generic[DataT]):
     pagination_type: PaginationType | None = None
     filter_attributes: dict[str, list[Any]] | None = None
 
+    _discriminated_union_cache: ClassVar[dict[Any, Any]] = {}
+
+    def __class_getitem__(  # type: ignore[invalid-method-override]
+        cls, item: type[Any] | tuple[type[Any], ...]
+    ) -> type[Any]:
+        if cls is PaginatedResponse and not isinstance(item, TypeVar):
+            cached = cls._discriminated_union_cache.get(item)
+            if cached is None:
+                cached = Annotated[
+                    Union[CursorPaginatedResponse[item], OffsetPaginatedResponse[item]],  # type: ignore[invalid-type-form]
+                    Field(discriminator="pagination_type"),
+                ]
+                cls._discriminated_union_cache[item] = cached
+            return cached  # type: ignore[invalid-return-type]
+        return super().__class_getitem__(item)
+
 
 class OffsetPaginatedResponse(PaginatedResponse[DataT]):
     """Paginated response with typed offset-based pagination metadata.

tests/test_crud.py

@@ -7,7 +7,7 @@ from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
 from fastapi_toolsets.crud import CrudFactory, PaginationType
-from fastapi_toolsets.crud.factory import AsyncCrud
+from fastapi_toolsets.crud.factory import AsyncCrud, _CursorDirection
 from fastapi_toolsets.exceptions import NotFoundError
 
 from .conftest import (
@@ -2016,7 +2016,7 @@ class TestCursorPaginatePrevCursor:
         assert isinstance(page1.pagination, CursorPagination)
 
         # Manually craft a backward cursor before any existing id
-        before_all = _encode_cursor(0, direction="prev")
+        before_all = _encode_cursor(0, direction=_CursorDirection.PREV)
         empty = await IntRoleCursorCrud.cursor_paginate(
             db_session, cursor=before_all, items_per_page=5, schema=IntRoleRead
         )

tests/test_schemas.py

@@ -304,7 +304,7 @@ class TestPaginatedResponse:
             page=1,
             has_more=False,
         )
-        response = PaginatedResponse[dict](
+        response = PaginatedResponse(
             data=[],
             pagination=pagination,
         )
@@ -313,6 +313,30 @@ class TestPaginatedResponse:
         assert response.data == []
         assert response.pagination.total_count == 0
 
+    def test_class_getitem_with_concrete_type_returns_discriminated_union(self):
+        """PaginatedResponse[T] with a concrete type returns a discriminated Annotated union."""
+        import typing
+
+        alias = PaginatedResponse[dict]
+        args = typing.get_args(alias)
+        # args[0] is the Union, args[1] is the FieldInfo discriminator
+        union_args = typing.get_args(args[0])
+        assert CursorPaginatedResponse[dict] in union_args
+        assert OffsetPaginatedResponse[dict] in union_args
+
+    def test_class_getitem_is_cached(self):
+        """Repeated subscripting with the same type returns the identical cached object."""
+        assert PaginatedResponse[dict] is PaginatedResponse[dict]
+
+    def test_class_getitem_with_typevar_returns_generic(self):
+        """PaginatedResponse[TypeVar] falls through to Pydantic generic parametrisation."""
+        from typing import TypeVar
+
+        T = TypeVar("T")
+        alias = PaginatedResponse[T]
+        # Should be a generic alias, not an Annotated union
+        assert not hasattr(alias, "__metadata__")
+
     def test_generic_type_hint(self):
         """PaginatedResponse supports generic type hints."""
         pagination = OffsetPagination(

uv.lock

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