docs: add authentication example

* feat: unified paginate() endpoint with typed pagination responses

* docs: unified paginate() endpoint

* fix: add tests

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`)
-- **Standardized API Responses**: Consistent response format with `Response`, `PaginatedResponse`, and `PydanticBase`
+- **Model Mixins**: SQLAlchemy mixins for common column patterns (`UUIDMixin`, `UUIDv7Mixin`, `CreatedAtMixin`, `UpdatedAtMixin`, `TimestampMixin`)
+- **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/examples/authentication.md

@@ -0,0 +1 @@
+# Authentication

docs/examples/pagination-search.md

@@ -42,12 +42,17 @@ Declare `searchable_fields`, `facet_fields`, and `order_fields` once on [`CrudFa
 
 
 ## Routes
+
+```python title="routes.py:1:17"
+--8<-- "docs_src/examples/pagination_search/routes.py:1:17"
+```
+
 ### Offset pagination
 
 Best for admin panels or any UI that needs a total item count and numbered pages.
 
-```python title="routes.py:1:36"
---8<-- "docs_src/examples/pagination_search/routes.py:1:36"
+```python title="routes.py:20:40"
+--8<-- "docs_src/examples/pagination_search/routes.py:20:40"
 ```
 
 **Example request**
@@ -61,6 +66,7 @@ GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published&or
 ```json
 {
   "status": "SUCCESS",
+  "pagination_type": "offset",
   "data": [
     { "id": "3f47ac69-...", "title": "FastAPI tips", "status": "published", ... }
   ],
@@ -83,8 +89,8 @@ GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published&or
 
 Best for feeds, infinite scroll, or any high-throughput API where offset performance degrades.
 
-```python title="routes.py:39:59"
---8<-- "docs_src/examples/pagination_search/routes.py:39:59"
+```python title="routes.py:43:63"
+--8<-- "docs_src/examples/pagination_search/routes.py:43:63"
 ```
 
 **Example request**
@@ -98,6 +104,7 @@ GET /articles/cursor?items_per_page=10&status=published&order_by=created_at&orde
 ```json
 {
   "status": "SUCCESS",
+  "pagination_type": "cursor",
   "data": [
     { "id": "3f47ac69-...", "title": "FastAPI tips", "status": "published", ... }
   ],
@@ -116,6 +123,47 @@ GET /articles/cursor?items_per_page=10&status=published&order_by=created_at&orde
 
 Pass `next_cursor` as the `cursor` query parameter on the next request to advance to the next page.
 
+### Unified endpoint (both strategies)
+
+!!! info "Added in `v2.3.0`"
+
+[`paginate()`](../module/crud.md#unified-paginate--both-strategies-on-one-endpoint) lets a single endpoint support both strategies via a `pagination_type` query parameter. The `pagination_type` field in the response acts as a discriminator for frontend tooling.
+
+```python title="routes.py:66:90"
+--8<-- "docs_src/examples/pagination_search/routes.py:66:90"
+```
+
+**Offset request** (default)
+
+```
+GET /articles/?pagination_type=offset&page=1&items_per_page=10
+```
+
+```json
+{
+  "status": "SUCCESS",
+  "pagination_type": "offset",
+  "data": ["..."],
+  "pagination": { "total_count": 42, "page": 1, "items_per_page": 10, "has_more": true }
+}
+```
+
+**Cursor request**
+
+```
+GET /articles/?pagination_type=cursor&items_per_page=10
+GET /articles/?pagination_type=cursor&items_per_page=10&cursor=eyJ2YWx1ZSI6...
+```
+
+```json
+{
+  "status": "SUCCESS",
+  "pagination_type": "cursor",
+  "data": ["..."],
+  "pagination": { "next_cursor": "eyJ2YWx1ZSI6...", "prev_cursor": null, "items_per_page": 10, "has_more": true }
+}
+```
+
 ## Search behaviour
 
 Both endpoints inherit the same `searchable_fields` declared on `ArticleCrud`:

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`)
-- **Standardized API Responses**: Consistent response format with `Response`, `PaginatedResponse`, and `PydanticBase`
+- **Model Mixins**: SQLAlchemy mixins for common column patterns (`UUIDMixin`, `UUIDv7Mixin`, `CreatedAtMixin`, `UpdatedAtMixin`, `TimestampMixin`)
+- **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/crud.md

@@ -7,10 +7,12 @@ Generic async CRUD operations for SQLAlchemy models with search, pagination, and
 
 ## Overview
 
-The `crud` module provides [`AsyncCrud`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud), an abstract base class with a full suite of async database operations, and [`CrudFactory`](../reference/crud.md#fastapi_toolsets.crud.factory.CrudFactory), a convenience function to instantiate it for a given model.
+The `crud` module provides [`AsyncCrud`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud), a base class with a full suite of async database operations, and [`CrudFactory`](../reference/crud.md#fastapi_toolsets.crud.factory.CrudFactory), a convenience function to instantiate it for a given model.
 
 ## Creating a CRUD class
 
+### Factory style
+
 ```python
 from fastapi_toolsets.crud import CrudFactory
 from myapp.models import User
@@ -18,7 +20,65 @@ from myapp.models import User
 UserCrud = CrudFactory(model=User)
 ```
 
-[`CrudFactory`](../reference/crud.md#fastapi_toolsets.crud.factory.CrudFactory) dynamically creates a class named `AsyncUserCrud` with `User` as its model.
+[`CrudFactory`](../reference/crud.md#fastapi_toolsets.crud.factory.CrudFactory) dynamically creates a class named `AsyncUserCrud` with `User` as its model. This is the most concise option for straightforward CRUD with no custom logic.
+
+### Subclass style
+
+!!! info "Added in `v2.3.0`"
+
+```python
+from fastapi_toolsets.crud.factory import AsyncCrud
+from myapp.models import User
+
+class UserCrud(AsyncCrud[User]):
+    model = User
+    searchable_fields = [User.username, User.email]
+    default_load_options = [selectinload(User.role)]
+```
+
+Subclassing [`AsyncCrud`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud) directly is the preferred style when you need to add custom methods or when the configuration is complex enough to benefit from a named class body. 
+
+### Adding custom methods
+
+```python
+class UserCrud(AsyncCrud[User]):
+    model = User
+
+    @classmethod
+    async def get_active(cls, session: AsyncSession) -> list[User]:
+        return await cls.get_multi(session, filters=[User.is_active == True])
+```
+
+### Sharing a custom base across multiple models
+
+Define a generic base class with the shared methods, then subclass it for each model:
+
+```python
+from typing import Generic, TypeVar
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy.orm import DeclarativeBase
+from fastapi_toolsets.crud.factory import AsyncCrud
+
+T = TypeVar("T", bound=DeclarativeBase)
+
+class AuditedCrud(AsyncCrud[T], Generic[T]):
+    """Base CRUD with custom function"""
+
+    @classmethod
+    async def get_active(cls, session: AsyncSession):
+        return await cls.get_multi(session, filters=[cls.model.is_active == True])
+
+
+class UserCrud(AuditedCrud[User]):
+    model = User
+    searchable_fields = [User.username, User.email]
+```
+
+You can also use the factory shorthand with the same base by passing `base_class`:
+
+```python
+UserCrud = CrudFactory(User, base_class=AuditedCrud)
+```
 
 ## Basic operations
 
@@ -85,41 +145,40 @@ user = await UserCrud.first(session=session, filters=[User.is_active == True])
 
 !!! info "Added in `v1.1` (only offset_pagination via `paginate` if `<v1.1`)"
 
-Two pagination strategies are available. Both return a [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse) but differ in how they navigate through results.
+Three pagination methods are available.  All return a typed response whose `pagination_type` field tells clients which strategy was used.
 
-| | `offset_paginate` | `cursor_paginate` |
-|---|---|---|
-| Total count | Yes | No |
-| Jump to arbitrary page | Yes | No |
-| Performance on deep pages | Degrades | Constant |
-| Stable under concurrent inserts | No | Yes |
-| Search compatible | Yes | Yes |
-| Use case | Admin panels, numbered pagination | Feeds, APIs, infinite scroll |
+| | `offset_paginate` | `cursor_paginate` | `paginate` |
+|---|---|---|---|
+| Return type | `OffsetPaginatedResponse` | `CursorPaginatedResponse` | either, based on `pagination_type` param |
+| Total count | Yes | No | / |
+| Jump to arbitrary page | Yes | No | / |
+| Performance on deep pages | Degrades | Constant | / |
+| Stable under concurrent inserts | No | Yes | / |
+| Use case | Admin panels, numbered pagination | Feeds, APIs, infinite scroll | single endpoint, both strategies |
 
 ### Offset pagination
 
 ```python
-@router.get(
-    "",
-    response_model=PaginatedResponse[User],
-)
+@router.get("")
 async def get_users(
     session: SessionDep,
     items_per_page: int = 50,
     page: int = 1,
-):
-    return await crud.UserCrud.offset_paginate(
+) -> OffsetPaginatedResponse[UserRead]:
+    return await UserCrud.offset_paginate(
         session=session,
         items_per_page=items_per_page,
         page=page,
+        schema=UserRead,
     )
 ```
 
-The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) method returns a [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse) whose `pagination` field is an [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination) object:
+The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) method returns an [`OffsetPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPaginatedResponse):
 
 ```json
 {
   "status": "SUCCESS",
+  "pagination_type": "offset",
   "data": ["..."],
   "pagination": {
     "total_count": 100,
@@ -133,27 +192,26 @@ The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.Async
 ### Cursor pagination
 
 ```python
-@router.get(
-    "",
-    response_model=PaginatedResponse[UserRead],
-)
+@router.get("")
 async def list_users(
     session: SessionDep,
     cursor: str | None = None,
     items_per_page: int = 20,
-):
+) -> CursorPaginatedResponse[UserRead]:
     return await UserCrud.cursor_paginate(
         session=session,
         cursor=cursor,
         items_per_page=items_per_page,
+        schema=UserRead,
     )
 ```
 
-The [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate) method returns a [`PaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse) whose `pagination` field is a [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination) object:
+The [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate) method returns a [`CursorPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPaginatedResponse):
 
 ```json
 {
   "status": "SUCCESS",
+  "pagination_type": "cursor",
   "data": ["..."],
   "pagination": {
     "next_cursor": "eyJ2YWx1ZSI6ICIzZjQ3YWM2OS0uLi4ifQ==",
@@ -198,6 +256,41 @@ PostCrud = CrudFactory(model=Post, cursor_column=Post.id)
 PostCrud = CrudFactory(model=Post, cursor_column=Post.created_at)
 ```
 
+### Unified endpoint (both strategies)
+
+!!! info "Added in `v2.3.0`"
+
+[`paginate()`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.paginate) dispatches to `offset_paginate` or `cursor_paginate` based on a `pagination_type` query parameter, letting you expose **one endpoint** that supports both strategies.  The `pagination_type` field in the response tells clients which strategy was used, enabling frontend discriminated-union typing.
+
+```python
+from fastapi_toolsets.crud import PaginationType
+from fastapi_toolsets.schemas import PaginatedResponse
+
+@router.get("")
+async def list_users(
+    session: SessionDep,
+    pagination_type: PaginationType = PaginationType.OFFSET,
+    page: int = Query(1, ge=1, description="Current page (offset only)"),
+    cursor: str | None = Query(None, description="Cursor token (cursor only)"),
+    items_per_page: int = Query(20, ge=1, le=100),
+) -> PaginatedResponse[UserRead]:
+    return await UserCrud.paginate(
+        session,
+        pagination_type=pagination_type,
+        page=page,
+        cursor=cursor,
+        items_per_page=items_per_page,
+        schema=UserRead,
+    )
+```
+
+```
+GET /users?pagination_type=offset&page=2&items_per_page=10
+GET /users?pagination_type=cursor&cursor=eyJ2YWx1ZSI6...&items_per_page=10
+```
+
+Both `page` and `cursor` are always accepted by the endpoint — unused parameters are silently ignored by `paginate()`.
+
 ## Search
 
 Two search strategies are available, both compatible with [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) and [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate).
@@ -240,40 +333,36 @@ result = await UserCrud.offset_paginate(
 This allows searching with both [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) and [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate):
 
 ```python
-@router.get(
-    "",
-    response_model=PaginatedResponse[User],
-)
+@router.get("")
 async def get_users(
     session: SessionDep,
     items_per_page: int = 50,
     page: int = 1,
     search: str | None = None,
-):
-    return await crud.UserCrud.offset_paginate(
+) -> OffsetPaginatedResponse[UserRead]:
+    return await UserCrud.offset_paginate(
         session=session,
         items_per_page=items_per_page,
         page=page,
         search=search,
+        schema=UserRead,
     )
 ```
 
 ```python
-@router.get(
-    "",
-    response_model=PaginatedResponse[User],
-)
+@router.get("")
 async def get_users(
     session: SessionDep,
     cursor: str | None = None,
     items_per_page: int = 50,
     search: str | None = None,
-):
-    return await crud.UserCrud.cursor_paginate(
+) -> CursorPaginatedResponse[UserRead]:
+    return await UserCrud.cursor_paginate(
         session=session,
         items_per_page=items_per_page,
         cursor=cursor,
         search=search,
+        schema=UserRead,
     )
 ```
 
@@ -344,11 +433,12 @@ async def list_users(
     session: SessionDep,
     page: int = 1,
     filter_by: Annotated[dict[str, list[str]], Depends(UserCrud.filter_params())],
-) -> PaginatedResponse[UserRead]:
+) -> OffsetPaginatedResponse[UserRead]:
     return await UserCrud.offset_paginate(
         session=session,
         page=page,
         filter_by=filter_by,
+        schema=UserRead,
     )
 ```
 
@@ -388,8 +478,8 @@ from fastapi_toolsets.crud import OrderByClause
 async def list_users(
     session: SessionDep,
     order_by: Annotated[OrderByClause | None, Depends(UserCrud.order_params())],
-) -> PaginatedResponse[UserRead]:
-    return await UserCrud.offset_paginate(session=session, order_by=order_by)
+) -> OffsetPaginatedResponse[UserRead]:
+    return await UserCrud.offset_paginate(session=session, order_by=order_by, schema=UserRead)
 ```
 
 The dependency adds two query parameters to the endpoint:
@@ -496,7 +586,7 @@ async def get_user(session: SessionDep, uuid: UUID) -> Response[UserRead]:
     )
 
 @router.get("")
-async def list_users(session: SessionDep, page: int = 1) -> PaginatedResponse[UserRead]:
+async def list_users(session: SessionDep, page: int = 1) -> OffsetPaginatedResponse[UserRead]:
     return await crud.UserCrud.offset_paginate(
         session=session,
         page=page,

docs/module/models.md

@@ -18,13 +18,15 @@ class Article(Base, UUIDMixin, TimestampMixin):
     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.
+All timestamp columns are timezone-aware (`TIMESTAMPTZ`). All defaults are server-side (`clock_timestamp()`), 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()`.
+Adds a `id: UUID` primary key generated server-side by PostgreSQL using `gen_random_uuid()`. The value is retrieved via `RETURNING` after insert, so it is available on the Python object immediately after `flush()`.
+
+!!! warning "Requires PostgreSQL 13+"
 
 ```python
 from fastapi_toolsets.models import UUIDMixin
@@ -36,13 +38,37 @@ class User(Base, UUIDMixin):
 
 # id is None before flush
 user = User(username="alice")
+session.add(user)
 await session.flush()
 print(user.id)  # UUID('...')
 ```
 
+### [`UUIDv7Mixin`](../reference/models.md#fastapi_toolsets.models.UUIDv7Mixin)
+
+!!! info "Added in `v2.3`"
+
+Adds a `id: UUID` primary key generated server-side by PostgreSQL using `uuidv7()`. It's a time-ordered UUID format that encodes a millisecond-precision timestamp in the most significant bits, making it naturally sortable and index-friendly.
+
+!!! warning "Requires PostgreSQL 18+"
+
+```python
+from fastapi_toolsets.models import UUIDv7Mixin
+
+class Event(Base, UUIDv7Mixin):
+    __tablename__ = "events"
+
+    name: Mapped[str]
+
+# id is None before flush
+event = Event(name="user.signup")
+session.add(event)
+await session.flush()
+print(event.id)  # UUID('019...')
+```
+
 ### [`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.
+Adds a `created_at: datetime` column set to `clock_timestamp()` 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
@@ -55,7 +81,7 @@ class Order(Base, UUIDMixin, CreatedAtMixin):
 
 ### [`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).
+Adds an `updated_at: datetime` column set to `clock_timestamp()` on insert and automatically updated to `clock_timestamp()` on every ORM-level update (via SQLAlchemy's `onupdate` hook).
 
 ```python
 from fastapi_toolsets.models import UUIDMixin, UpdatedAtMixin

docs/module/schemas.md

@@ -20,50 +20,115 @@ async def get_user(user: User = UserDep) -> Response[UserSchema]:
     return Response(data=user, message="User retrieved")
 ```
 
-### [`PaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse)
+### Paginated response models
 
-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.
+Three classes wrap paginated list results. Pick the one that matches your endpoint's strategy:
 
-#### [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination)
+| Class | `pagination` type | `pagination_type` field | Use when |
+|---|---|---|---|
+| [`OffsetPaginatedResponse[T]`](#offsetpaginatedresponset) | `OffsetPagination` | `"offset"` (fixed) | endpoint always uses offset |
+| [`CursorPaginatedResponse[T]`](#cursorpaginatedresponset) | `CursorPagination` | `"cursor"` (fixed) | endpoint always uses cursor |
+| [`PaginatedResponse[T]`](#paginatedresponset) | `OffsetPagination \| CursorPagination` | — | unified endpoint supporting both strategies |
 
-Page-number based. Requires `total_count` so clients can compute the total number of pages.
+#### [`OffsetPaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPaginatedResponse)
+
+!!! info "Added in `v2.3.0`"
+
+Use as the return type when the endpoint always uses [`offset_paginate`](crud.md#offset-pagination).  The `pagination` field is guaranteed to be an [`OffsetPagination`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPagination) object; the response always includes a `pagination_type: "offset"` discriminator.
 
 ```python
-from fastapi_toolsets.schemas import PaginatedResponse, OffsetPagination
+from fastapi_toolsets.schemas import OffsetPaginatedResponse
 
 @router.get("/users")
-async def list_users() -> PaginatedResponse[UserSchema]:
-    return PaginatedResponse(
-        data=users,
-        pagination=OffsetPagination(
-            total_count=100,
-            items_per_page=10,
-            page=1,
-            has_more=True,
-        ),
+async def list_users(
+    page: int = 1,
+    items_per_page: int = 20,
+) -> OffsetPaginatedResponse[UserSchema]:
+    return await UserCrud.offset_paginate(
+        session, page=page, items_per_page=items_per_page, schema=UserSchema
     )
 ```
 
-#### [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination)
+**Response shape:**
 
-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.
+```json
+{
+  "status": "SUCCESS",
+  "pagination_type": "offset",
+  "data": ["..."],
+  "pagination": {
+    "total_count": 100,
+    "page": 1,
+    "items_per_page": 20,
+    "has_more": true
+  }
+}
+```
+
+#### [`CursorPaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPaginatedResponse)
+
+!!! info "Added in `v2.3.0`"
+
+Use as the return type when the endpoint always uses [`cursor_paginate`](crud.md#cursor-pagination).  The `pagination` field is guaranteed to be a [`CursorPagination`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPagination) object; the response always includes a `pagination_type: "cursor"` discriminator.
 
 ```python
-from fastapi_toolsets.schemas import PaginatedResponse, CursorPagination
+from fastapi_toolsets.schemas import CursorPaginatedResponse
 
 @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,
-        ),
+async def list_events(
+    cursor: str | None = None,
+    items_per_page: int = 20,
+) -> CursorPaginatedResponse[EventSchema]:
+    return await EventCrud.cursor_paginate(
+        session, cursor=cursor, items_per_page=items_per_page, schema=EventSchema
     )
 ```
 
+**Response shape:**
+
+```json
+{
+  "status": "SUCCESS",
+  "pagination_type": "cursor",
+  "data": ["..."],
+  "pagination": {
+    "next_cursor": "eyJpZCI6IDQyfQ==",
+    "prev_cursor": null,
+    "items_per_page": 20,
+    "has_more": true
+  }
+}
+```
+
+#### [`PaginatedResponse[T]`](../reference/schemas.md#fastapi_toolsets.schemas.PaginatedResponse)
+
+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
+from fastapi_toolsets.schemas import PaginatedResponse
+
+@router.get("/users")
+async def list_users(
+    pagination_type: PaginationType = PaginationType.OFFSET,
+    page: int = 1,
+    cursor: str | None = None,
+    items_per_page: int = 20,
+) -> PaginatedResponse[UserSchema]:
+    return await UserCrud.paginate(
+        session,
+        pagination_type=pagination_type,
+        page=page,
+        cursor=cursor,
+        items_per_page=items_per_page,
+        schema=UserSchema,
+    )
+```
+
+#### Pagination metadata models
+
 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)

docs/module/security.md

@@ -0,0 +1,267 @@
+# Security
+
+Composable authentication helpers for FastAPI that use `Security()` for OpenAPI documentation and accept user-provided validator functions with full type flexibility.
+
+## Overview
+
+The `security` module provides four auth source classes and a `MultiAuth` factory. Each class wraps a FastAPI security scheme for OpenAPI and accepts a validator function called as:
+
+```python
+await validator(credential, **kwargs)
+```
+
+where `kwargs` are the extra keyword arguments provided at instantiation (roles, permissions, enums, etc.). The validator returns the authenticated identity (e.g. a `User` model) which becomes the route dependency value.
+
+```python
+from fastapi import Security
+from fastapi_toolsets.security import BearerTokenAuth
+
+async def verify_token(token: str, *, role: str) -> User:
+    user = await db.get_by_token(token)
+    if not user or user.role != role:
+        raise UnauthorizedError()
+    return user
+
+bearer_admin = BearerTokenAuth(verify_token, role="admin")
+
+@app.get("/admin")
+async def admin_route(user: User = Security(bearer_admin)):
+    return user
+```
+
+## Auth sources
+
+### [`BearerTokenAuth`](../reference/security.md#fastapi_toolsets.security.BearerTokenAuth)
+
+Reads the `Authorization: Bearer <token>` header. Wraps `HTTPBearer` for OpenAPI.
+
+```python
+from fastapi_toolsets.security import BearerTokenAuth
+
+bearer = BearerTokenAuth(validator=verify_token)
+
+@app.get("/me")
+async def me(user: User = Security(bearer)):
+    return user
+```
+
+#### Token prefix
+
+The optional `prefix` parameter restricts a `BearerTokenAuth` instance to tokens
+that start with a given string. The prefix is **kept** in the value passed to the
+validator — store and compare tokens with their prefix included.
+
+This lets you deploy multiple `BearerTokenAuth` instances in the same application
+and disambiguate them efficiently in `MultiAuth`:
+
+```python
+user_bearer = BearerTokenAuth(verify_user, prefix="user_")  # matches "Bearer user_..."
+org_bearer  = BearerTokenAuth(verify_org,  prefix="org_")   # matches "Bearer org_..."
+```
+
+Use [`generate_token()`](#token-generation) to create correctly-prefixed tokens.
+
+#### Token generation
+
+`BearerTokenAuth.generate_token()` produces a secure random token ready to store
+in your database and return to the client. If a prefix is configured it is
+prepended automatically:
+
+```python
+bearer = BearerTokenAuth(verify_token, prefix="user_")
+
+token = bearer.generate_token()   # e.g. "user_Xk3mN..."
+await db.store_token(user_id, token)
+return {"access_token": token, "token_type": "bearer"}
+```
+
+The client sends `Authorization: Bearer user_Xk3mN...` and the validator receives
+the full token (prefix included) to compare against the stored value.
+
+### [`CookieAuth`](../reference/security.md#fastapi_toolsets.security.CookieAuth)
+
+Reads a named cookie. Wraps `APIKeyCookie` for OpenAPI.
+
+```python
+from fastapi_toolsets.security import CookieAuth
+
+cookie_auth = CookieAuth("session", validator=verify_session)
+
+@app.get("/me")
+async def me(user: User = Security(cookie_auth)):
+    return user
+```
+
+### [`OAuth2Auth`](../reference/security.md#fastapi_toolsets.security.OAuth2Auth)
+
+Reads the `Authorization: Bearer <token>` header and registers the token endpoint
+in OpenAPI via `OAuth2PasswordBearer`.
+
+```python
+from fastapi_toolsets.security import OAuth2Auth
+
+oauth2_auth = OAuth2Auth(token_url="/token", validator=verify_token)
+
+@app.get("/me")
+async def me(user: User = Security(oauth2_auth)):
+    return user
+```
+
+### [`OpenIDAuth`](../reference/security.md#fastapi_toolsets.security.OpenIDAuth)
+
+Reads the `Authorization: Bearer <token>` header and registers the OpenID Connect
+discovery URL in OpenAPI via `OpenIdConnect`. Token validation is fully delegated
+to your validator — use any OIDC / JWT library (`authlib`, `python-jose`, `PyJWT`).
+
+```python
+from fastapi_toolsets.security import OpenIDAuth
+
+async def verify_google_token(token: str, *, audience: str) -> User:
+    payload = jwt.decode(token, google_public_keys, algorithms=["RS256"],
+                         audience=audience)
+    return User(email=payload["email"], name=payload["name"])
+
+google_auth = OpenIDAuth(
+    "https://accounts.google.com/.well-known/openid-configuration",
+    verify_google_token,
+    audience="my-client-id",
+)
+
+@app.get("/me")
+async def me(user: User = Security(google_auth)):
+    return user
+```
+
+The discovery URL is used **only for OpenAPI documentation** — no requests are made
+to it by this class. You are responsible for fetching and caching the provider's
+public keys in your validator.
+
+Multiple providers work naturally with `MultiAuth`:
+
+```python
+multi = MultiAuth(google_auth, github_auth)
+
+@app.get("/data")
+async def data(user: User = Security(multi)):
+    return user
+```
+
+## Typed validator kwargs
+
+All auth classes forward extra instantiation keyword arguments to the validator.
+Arguments can be any type — enums, strings, integers, etc. The validator returns
+the authenticated identity, which FastAPI injects directly into the route handler.
+
+```python
+async def verify_token(token: str, *, role: Role, permission: str) -> User:
+    user = await decode_token(token)
+    if user.role != role or permission not in user.permissions:
+        raise UnauthorizedError()
+    return user
+
+bearer = BearerTokenAuth(verify_token, role=Role.ADMIN, permission="billing:read")
+```
+
+Each auth instance is self-contained — create a separate instance per distinct
+requirement instead of passing requirements through `Security(scopes=[...])`.
+
+### Using `.require()` inline
+
+If declaring a new top-level variable per role feels verbose, use `.require()` to
+create a configured clone directly in the route decorator. The original instance
+is not mutated:
+
+```python
+bearer = BearerTokenAuth(verify_token)
+
+@app.get("/admin/stats")
+async def admin_stats(user: User = Security(bearer.require(role=Role.ADMIN))):
+    return {"message": f"Hello admin {user.name}"}
+
+@app.get("/profile")
+async def profile(user: User = Security(bearer.require(role=Role.USER))):
+    return {"id": user.id, "name": user.name}
+```
+
+`.require()` kwargs are merged over existing ones — new values win on conflict.
+The `prefix` (for `BearerTokenAuth`) and cookie name (for `CookieAuth`) are
+always preserved.
+
+`.require()` instances work transparently inside `MultiAuth`:
+
+```python
+multi = MultiAuth(
+    user_bearer.require(role=Role.USER),
+    org_bearer.require(role=Role.ADMIN),
+)
+```
+
+## MultiAuth
+
+[`MultiAuth`](../reference/security.md#fastapi_toolsets.security.MultiAuth) combines
+multiple auth sources into a single callable. Sources are tried in order; the
+first one that finds a credential wins.
+
+```python
+from fastapi_toolsets.security import MultiAuth
+
+multi = MultiAuth(user_bearer, org_bearer, cookie_auth)
+
+@app.get("/data")
+async def data_route(user = Security(multi)):
+    return user
+```
+
+### Using `.require()` on MultiAuth
+
+`MultiAuth` also supports `.require()`, which propagates the kwargs to every
+source that implements it. Sources that do not (e.g. custom `AuthSource`
+subclasses) are passed through unchanged:
+
+```python
+multi = MultiAuth(bearer, cookie)
+
+@app.get("/admin")
+async def admin(user: User = Security(multi.require(role=Role.ADMIN))):
+    return user
+```
+
+This is equivalent to calling `.require()` on each source individually:
+
+```python
+# These two are identical
+multi.require(role=Role.ADMIN)
+
+MultiAuth(
+    bearer.require(role=Role.ADMIN),
+    cookie.require(role=Role.ADMIN),
+)
+```
+
+### Prefix-based dispatch
+
+Because `extract()` is pure string matching (no I/O), prefix-based source
+selection is essentially free. Only the matching source's validator (which may
+involve DB or network I/O) is ever called:
+
+```python
+user_bearer = BearerTokenAuth(verify_user, prefix="user_")
+org_bearer  = BearerTokenAuth(verify_org,  prefix="org_")
+
+multi = MultiAuth(user_bearer, org_bearer)
+
+# "Bearer user_alice" → only verify_user runs, receives "user_alice"
+# "Bearer org_acme"   → only verify_org runs, receives "org_acme"
+```
+
+Tokens are stored and compared **with their prefix** — use `generate_token()` on
+each source to issue correctly-prefixed tokens:
+
+```python
+user_token = user_bearer.generate_token()  # "user_..."
+org_token  = org_bearer.generate_token()   # "org_..."
+```
+
+---
+
+[:material-api: API Reference](../reference/security.md)

docs/reference/models.md

@@ -7,6 +7,7 @@ You can import them directly from `fastapi_toolsets.models`:
 ```python
 from fastapi_toolsets.models import (
     UUIDMixin,
+    UUIDv7Mixin,
     CreatedAtMixin,
     UpdatedAtMixin,
     TimestampMixin,
@@ -15,6 +16,8 @@ from fastapi_toolsets.models import (
 
 ## ::: fastapi_toolsets.models.UUIDMixin
 
+## ::: fastapi_toolsets.models.UUIDv7Mixin
+
 ## ::: fastapi_toolsets.models.CreatedAtMixin
 
 ## ::: fastapi_toolsets.models.UpdatedAtMixin

docs/reference/schemas.md

@@ -14,7 +14,10 @@ from fastapi_toolsets.schemas import (
     ErrorResponse,
     OffsetPagination,
     CursorPagination,
+    PaginationType,
     PaginatedResponse,
+    OffsetPaginatedResponse,
+    CursorPaginatedResponse,
 )
 ```
 
@@ -34,4 +37,10 @@ from fastapi_toolsets.schemas import (
 
 ## ::: fastapi_toolsets.schemas.CursorPagination
 
+## ::: fastapi_toolsets.schemas.PaginationType
+
 ## ::: fastapi_toolsets.schemas.PaginatedResponse
+
+## ::: fastapi_toolsets.schemas.OffsetPaginatedResponse
+
+## ::: fastapi_toolsets.schemas.CursorPaginatedResponse

docs/reference/security.md

@@ -0,0 +1,28 @@
+# `security`
+
+Here's the reference for the authentication helpers provided by the `security` module.
+
+You can import them directly from `fastapi_toolsets.security`:
+
+```python
+from fastapi_toolsets.security import (
+    AuthSource,
+    BearerTokenAuth,
+    CookieAuth,
+    OAuth2Auth,
+    OpenIDAuth,
+    MultiAuth,
+)
+```
+
+## ::: fastapi_toolsets.security.AuthSource
+
+## ::: fastapi_toolsets.security.BearerTokenAuth
+
+## ::: fastapi_toolsets.security.CookieAuth
+
+## ::: fastapi_toolsets.security.OAuth2Auth
+
+## ::: fastapi_toolsets.security.OpenIDAuth
+
+## ::: fastapi_toolsets.security.MultiAuth

docs_src/examples/authentication/app.py

@@ -0,0 +1,9 @@
+from fastapi import FastAPI
+
+from fastapi_toolsets.exceptions import init_exceptions_handlers
+
+from .routes import router
+
+app = FastAPI()
+init_exceptions_handlers(app=app)
+app.include_router(router=router)

docs_src/examples/authentication/crud.py

@@ -0,0 +1,9 @@
+from fastapi_toolsets.crud import CrudFactory
+
+from .models import OAuthAccount, OAuthProvider, Team, User, UserToken
+
+TeamCrud = CrudFactory(model=Team)
+UserCrud = CrudFactory(model=User)
+UserTokenCrud = CrudFactory(model=UserToken)
+OAuthProviderCrud = CrudFactory(model=OAuthProvider)
+OAuthAccountCrud = CrudFactory(model=OAuthAccount)

docs_src/examples/authentication/db.py

@@ -0,0 +1,15 @@
+from fastapi import Depends
+from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
+
+from fastapi_toolsets.db import create_db_context, create_db_dependency
+
+DATABASE_URL = "postgresql+asyncpg://postgres:postgres@localhost:5432/postgres"
+
+engine = create_async_engine(url=DATABASE_URL, future=True)
+async_session_maker = async_sessionmaker(bind=engine, expire_on_commit=False)
+
+get_db = create_db_dependency(session_maker=async_session_maker)
+get_db_context = create_db_context(session_maker=async_session_maker)
+
+
+SessionDep = Depends(get_db)

docs_src/examples/authentication/models.py

@@ -0,0 +1,105 @@
+import enum
+from datetime import datetime
+from uuid import UUID
+
+from sqlalchemy import (
+    Boolean,
+    DateTime,
+    Enum,
+    ForeignKey,
+    Integer,
+    String,
+    UniqueConstraint,
+)
+from sqlalchemy.dialects.postgresql import UUID as PG_UUID
+from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
+
+from fastapi_toolsets.models import TimestampMixin, UUIDMixin
+
+
+class Base(DeclarativeBase, UUIDMixin):
+    type_annotation_map = {
+        str: String(),
+        int: Integer(),
+        UUID: PG_UUID(as_uuid=True),
+        datetime: DateTime(timezone=True),
+    }
+
+
+class UserRole(enum.Enum):
+    admin = "admin"
+    moderator = "moderator"
+    user = "user"
+
+
+class Team(Base, TimestampMixin):
+    __tablename__ = "teams"
+
+    name: Mapped[str] = mapped_column(String, unique=True, index=True)
+    users: Mapped[list["User"]] = relationship(back_populates="team")
+
+
+class User(Base, TimestampMixin):
+    __tablename__ = "users"
+
+    username: Mapped[str] = mapped_column(String, unique=True, index=True)
+    email: Mapped[str | None] = mapped_column(
+        String, unique=True, index=True, nullable=True
+    )
+    hashed_password: Mapped[str | None] = mapped_column(String, nullable=True)
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+    role: Mapped[UserRole] = mapped_column(Enum(UserRole), default=UserRole.user)
+
+    team_id: Mapped[UUID | None] = mapped_column(ForeignKey("teams.id"), nullable=True)
+    team: Mapped["Team | None"] = relationship(back_populates="users")
+    oauth_accounts: Mapped[list["OAuthAccount"]] = relationship(back_populates="user")
+    tokens: Mapped[list["UserToken"]] = relationship(back_populates="user")
+
+
+class UserToken(Base, TimestampMixin):
+    """API tokens for a user (multiple allowed)."""
+
+    __tablename__ = "user_tokens"
+
+    user_id: Mapped[UUID] = mapped_column(ForeignKey("users.id"))
+    # Store hashed token value
+    token_hash: Mapped[str] = mapped_column(String, unique=True, index=True)
+    name: Mapped[str | None] = mapped_column(String, nullable=True)
+    expires_at: Mapped[datetime | None] = mapped_column(
+        DateTime(timezone=True), nullable=True
+    )
+
+    user: Mapped["User"] = relationship(back_populates="tokens")
+
+
+class OAuthProvider(Base, TimestampMixin):
+    """Configurable OAuth2 / OpenID Connect provider."""
+
+    __tablename__ = "oauth_providers"
+
+    slug: Mapped[str] = mapped_column(String, unique=True, index=True)
+    name: Mapped[str] = mapped_column(String)
+    client_id: Mapped[str] = mapped_column(String)
+    client_secret: Mapped[str] = mapped_column(String)
+    discovery_url: Mapped[str] = mapped_column(String, nullable=False)
+    scopes: Mapped[str] = mapped_column(String, default="openid email profile")
+    is_active: Mapped[bool] = mapped_column(Boolean, default=True)
+
+    accounts: Mapped[list["OAuthAccount"]] = relationship(back_populates="provider")
+
+
+class OAuthAccount(Base, TimestampMixin):
+    """OAuth2 / OpenID Connect account linked to a user."""
+
+    __tablename__ = "oauth_accounts"
+    __table_args__ = (
+        UniqueConstraint("provider_id", "subject", name="uq_oauth_provider_subject"),
+    )
+
+    user_id: Mapped[UUID] = mapped_column(ForeignKey("users.id"))
+    provider_id: Mapped[UUID] = mapped_column(ForeignKey("oauth_providers.id"))
+    # OAuth `sub` / OpenID subject identifier
+    subject: Mapped[str] = mapped_column(String)
+
+    user: Mapped["User"] = relationship(back_populates="oauth_accounts")
+    provider: Mapped["OAuthProvider"] = relationship(back_populates="accounts")

docs_src/examples/authentication/routes.py

@@ -0,0 +1,122 @@
+from typing import Annotated
+from uuid import UUID
+
+import bcrypt
+from fastapi import APIRouter, Form, HTTPException, Response, Security
+
+from fastapi_toolsets.dependencies import PathDependency
+
+from .crud import UserCrud, UserTokenCrud
+from .db import SessionDep
+from .models import OAuthProvider, User, UserToken
+from .schemas import (
+    ApiTokenCreateRequest,
+    ApiTokenResponse,
+    RegisterRequest,
+    UserCreate,
+    UserResponse,
+)
+from .security import auth, cookie_auth, create_api_token
+
+ProviderDep = PathDependency(
+    model=OAuthProvider,
+    field=OAuthProvider.slug,
+    session_dep=SessionDep,
+    param_name="slug",
+)
+
+
+def hash_password(password: str) -> str:
+    return bcrypt.hashpw(password.encode(), bcrypt.gensalt()).decode()
+
+
+def verify_password(plain: str, hashed: str) -> bool:
+    return bcrypt.checkpw(plain.encode(), hashed.encode())
+
+
+router = APIRouter(prefix="/auth")
+
+
+@router.post("/register", response_model=UserResponse, status_code=201)
+async def register(body: RegisterRequest, session: SessionDep):
+    existing = await UserCrud.first(
+        session=session, filters=[User.username == body.username]
+    )
+    if existing:
+        raise HTTPException(status_code=409, detail="Username already taken")
+
+    user = await UserCrud.create(
+        session=session,
+        obj=UserCreate(
+            username=body.username,
+            email=body.email,
+            hashed_password=hash_password(body.password),
+        ),
+    )
+    return user
+
+
+@router.post("/token", status_code=204)
+async def login(
+    session: SessionDep,
+    response: Response,
+    username: Annotated[str, Form()],
+    password: Annotated[str, Form()],
+):
+    user = await UserCrud.first(session=session, filters=[User.username == username])
+
+    if (
+        not user
+        or not user.hashed_password
+        or not verify_password(password, user.hashed_password)
+    ):
+        raise HTTPException(status_code=401, detail="Invalid credentials")
+
+    if not user.is_active:
+        raise HTTPException(status_code=403, detail="Account disabled")
+
+    cookie_auth.set_cookie(response, str(user.id))
+
+
+@router.post("/logout", status_code=204)
+async def logout(response: Response):
+    cookie_auth.delete_cookie(response)
+
+
+@router.get("/me", response_model=UserResponse)
+async def me(user: User = Security(auth)):
+    return user
+
+
+@router.post("/tokens", response_model=ApiTokenResponse, status_code=201)
+async def create_token(
+    body: ApiTokenCreateRequest,
+    user: User = Security(auth),
+):
+    raw, token_row = await create_api_token(
+        user.id, name=body.name, expires_at=body.expires_at
+    )
+    return ApiTokenResponse(
+        id=token_row.id,
+        name=token_row.name,
+        expires_at=token_row.expires_at,
+        created_at=token_row.created_at,
+        token=raw,
+    )
+
+
+@router.delete("/tokens/{token_id}", status_code=204)
+async def revoke_token(
+    session: SessionDep,
+    token_id: UUID,
+    user: User = Security(auth),
+):
+    if not await UserTokenCrud.first(
+        session=session,
+        filters=[UserToken.id == token_id, UserToken.user_id == user.id],
+    ):
+        raise HTTPException(status_code=404, detail="Token not found")
+    await UserTokenCrud.delete(
+        session=session,
+        filters=[UserToken.id == token_id, UserToken.user_id == user.id],
+    )

docs_src/examples/authentication/schemas.py

@@ -0,0 +1,64 @@
+from datetime import datetime
+from uuid import UUID
+
+from pydantic import EmailStr
+
+from fastapi_toolsets.schemas import PydanticBase
+
+
+class RegisterRequest(PydanticBase):
+    username: str
+    password: str
+    email: EmailStr | None = None
+
+
+class UserResponse(PydanticBase):
+    id: UUID
+    username: str
+    email: str | None
+    role: str
+    is_active: bool
+
+    model_config = {"from_attributes": True}
+
+
+class ApiTokenCreateRequest(PydanticBase):
+    name: str | None = None
+    expires_at: datetime | None = None
+
+
+class ApiTokenResponse(PydanticBase):
+    id: UUID
+    name: str | None
+    expires_at: datetime | None
+    created_at: datetime
+    # Only populated on creation
+    token: str | None = None
+
+    model_config = {"from_attributes": True}
+
+
+class OAuthProviderResponse(PydanticBase):
+    slug: str
+    name: str
+
+    model_config = {"from_attributes": True}
+
+
+class UserCreate(PydanticBase):
+    username: str
+    email: str | None = None
+    hashed_password: str | None = None
+
+
+class UserTokenCreate(PydanticBase):
+    user_id: UUID
+    token_hash: str
+    name: str | None = None
+    expires_at: datetime | None = None
+
+
+class OAuthAccountCreate(PydanticBase):
+    user_id: UUID
+    provider_id: UUID
+    subject: str

docs_src/examples/authentication/security.py

@@ -0,0 +1,100 @@
+import hashlib
+from datetime import datetime, timezone
+from uuid import UUID
+
+from fastapi import HTTPException
+from sqlalchemy.orm import selectinload
+
+from fastapi_toolsets.exceptions import UnauthorizedError
+from fastapi_toolsets.security import (
+    APIKeyHeaderAuth,
+    BearerTokenAuth,
+    CookieAuth,
+    MultiAuth,
+)
+
+from .crud import UserCrud, UserTokenCrud
+from .db import get_db_context
+from .models import User, UserRole, UserToken
+from .schemas import UserTokenCreate
+
+SESSION_COOKIE = "session"
+SECRET_KEY = "123456789"
+
+
+def _hash_token(token: str) -> str:
+    return hashlib.sha256(token.encode()).hexdigest()
+
+
+async def _verify_token(token: str, role: UserRole | None = None) -> User:
+    async with get_db_context() as db:
+        user_token = await UserTokenCrud.first(
+            session=db,
+            filters=[UserToken.token_hash == _hash_token(token)],
+            load_options=[selectinload(UserToken.user)],
+        )
+
+        if user_token is None or not user_token.user.is_active:
+            raise UnauthorizedError()
+
+        if user_token.expires_at and user_token.expires_at < datetime.now(timezone.utc):
+            raise UnauthorizedError()
+
+        user = user_token.user
+
+        if role is not None and user.role != role:
+            raise HTTPException(status_code=403, detail="Insufficient permissions")
+
+        return user
+
+
+async def _verify_cookie(user_id: str, role: UserRole | None = None) -> User:
+    async with get_db_context() as db:
+        user = await UserCrud.first(
+            session=db,
+            filters=[User.id == UUID(user_id)],
+        )
+
+    if not user or not user.is_active:
+        raise UnauthorizedError()
+
+    if role is not None and user.role != role:
+        raise HTTPException(status_code=403, detail="Insufficient permissions")
+
+    return user
+
+
+bearer_auth = BearerTokenAuth(
+    validator=_verify_token,
+    prefix="ctf_",
+)
+header_auth = APIKeyHeaderAuth(
+    name="X-API-Key",
+    validator=_verify_token,
+)
+cookie_auth = CookieAuth(
+    name=SESSION_COOKIE,
+    validator=_verify_cookie,
+    secret_key=SECRET_KEY,
+)
+auth = MultiAuth(bearer_auth, header_auth, cookie_auth)
+
+
+async def create_api_token(
+    user_id: UUID,
+    *,
+    name: str | None = None,
+    expires_at: datetime | None = None,
+) -> tuple[str, UserToken]:
+    raw = bearer_auth.generate_token()
+    async with get_db_context() as db:
+        token_row = await UserTokenCrud.create(
+            session=db,
+            obj=UserTokenCreate(
+                user_id=user_id,
+                token_hash=_hash_token(raw),
+                name=name,
+                expires_at=expires_at,
+            ),
+        )
+    return raw, token_row

docs_src/examples/pagination_search/models.py

@@ -1,9 +1,10 @@
-import datetime
 import uuid
 
-from sqlalchemy import Boolean, DateTime, ForeignKey, String, Text, func
+from sqlalchemy import Boolean, ForeignKey, String, Text
 from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column, relationship
 
+from fastapi_toolsets.models import CreatedAtMixin
+
 
 class Base(DeclarativeBase):
     pass
@@ -18,13 +19,10 @@ class Category(Base):
     articles: Mapped[list["Article"]] = relationship(back_populates="category")
 
 
-class Article(Base):
+class Article(Base, CreatedAtMixin):
     __tablename__ = "articles"
 
     id: Mapped[uuid.UUID] = mapped_column(primary_key=True, default=uuid.uuid4)
-    created_at: Mapped[datetime.datetime] = mapped_column(
-        DateTime(timezone=True), server_default=func.now()
-    )
     title: Mapped[str] = mapped_column(String(256))
     body: Mapped[str] = mapped_column(Text)
     status: Mapped[str] = mapped_column(String(32))

docs_src/examples/pagination_search/routes.py

@@ -2,8 +2,12 @@ from typing import Annotated
 
 from fastapi import APIRouter, Depends, Query
 
-from fastapi_toolsets.crud import OrderByClause
-from fastapi_toolsets.schemas import PaginatedResponse
+from fastapi_toolsets.crud import OrderByClause, PaginationType
+from fastapi_toolsets.schemas import (
+    CursorPaginatedResponse,
+    OffsetPaginatedResponse,
+    PaginatedResponse,
+)
 
 from .crud import ArticleCrud
 from .db import SessionDep
@@ -24,7 +28,7 @@ async def list_articles_offset(
     page: int = Query(1, ge=1),
     items_per_page: int = Query(20, ge=1, le=100),
     search: str | None = None,
-) -> PaginatedResponse[ArticleRead]:
+) -> OffsetPaginatedResponse[ArticleRead]:
     return await ArticleCrud.offset_paginate(
         session=session,
         page=page,
@@ -47,7 +51,7 @@ async def list_articles_cursor(
     cursor: str | None = None,
     items_per_page: int = Query(20, ge=1, le=100),
     search: str | None = None,
-) -> PaginatedResponse[ArticleRead]:
+) -> CursorPaginatedResponse[ArticleRead]:
     return await ArticleCrud.cursor_paginate(
         session=session,
         cursor=cursor,
@@ -57,3 +61,30 @@ async def list_articles_cursor(
         order_by=order_by,
         schema=ArticleRead,
     )
+
+
+@router.get("/")
+async def list_articles(
+    session: SessionDep,
+    filter_by: Annotated[dict[str, list[str]], Depends(ArticleCrud.filter_params())],
+    order_by: Annotated[
+        OrderByClause | None,
+        Depends(ArticleCrud.order_params(default_field=Article.created_at)),
+    ],
+    pagination_type: PaginationType = PaginationType.OFFSET,
+    page: int = Query(1, ge=1),
+    cursor: str | None = None,
+    items_per_page: int = Query(20, ge=1, le=100),
+    search: str | None = None,
+) -> PaginatedResponse[ArticleRead]:
+    return await ArticleCrud.paginate(
+        session,
+        pagination_type=pagination_type,
+        page=page,
+        cursor=cursor,
+        items_per_page=items_per_page,
+        search=search,
+        filter_by=filter_by or None,
+        order_by=order_by,
+        schema=ArticleRead,
+    )

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/__init__.py

@@ -1,6 +1,7 @@
 """Generic async CRUD operations for SQLAlchemy models."""
 
 from ..exceptions import InvalidFacetFilterError, NoSearchableFieldsError
+from ..schemas import PaginationType
 from ..types import (
     FacetFieldType,
     JoinType,
@@ -8,10 +9,11 @@ from ..types import (
     OrderByClause,
     SearchFieldType,
 )
-from .factory import CrudFactory
+from .factory import AsyncCrud, CrudFactory
 from .search import SearchConfig, get_searchable_fields
 
 __all__ = [
+    "AsyncCrud",
     "CrudFactory",
     "FacetFieldType",
     "get_searchable_fields",
@@ -20,6 +22,7 @@ __all__ = [
     "M2MFieldType",
     "NoSearchableFieldsError",
     "OrderByClause",
+    "PaginationType",
     "SearchConfig",
     "SearchFieldType",
 ]

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
@@ -23,7 +24,14 @@ from sqlalchemy.sql.roles import WhereHavingRole
 
 from ..db import get_transaction
 from ..exceptions import InvalidOrderFieldError, NotFoundError
-from ..schemas import CursorPagination, OffsetPagination, PaginatedResponse, Response
+from ..schemas import (
+    CursorPaginatedResponse,
+    CursorPagination,
+    OffsetPaginatedResponse,
+    OffsetPagination,
+    PaginationType,
+    Response,
+)
 from ..types import (
     FacetFieldType,
     JoinType,
@@ -42,14 +50,43 @@ from .search import (
 )
 
 
-def _encode_cursor(value: Any) -> str:
-    """Encode cursor column value as an base64 string."""
-    return base64.b64encode(json.dumps(str(value)).encode()).decode()
+class _CursorDirection(str, Enum):
+    NEXT = "next"
+    PREV = "prev"
 
 
-def _decode_cursor(cursor: str) -> str:
-    """Decode cursor base64 string."""
-    return json.loads(base64.b64decode(cursor.encode()).decode())
+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, _CursorDirection]:
+    """Decode a cursor base64 string into ``(raw_value, direction)``."""
+    payload = json.loads(base64.b64decode(cursor.encode()).decode())
+    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:
@@ -82,6 +119,27 @@ class AsyncCrud(Generic[ModelType]):
     default_load_options: ClassVar[Sequence[ExecutableOption] | None] = None
     cursor_column: ClassVar[Any | None] = None
 
+    @classmethod
+    def __init_subclass__(cls, **kwargs: Any) -> None:
+        super().__init_subclass__(**kwargs)
+        if "model" not in cls.__dict__:
+            return
+        model: type[DeclarativeBase] = cls.__dict__["model"]
+        pk_key = model.__mapper__.primary_key[0].key
+        assert pk_key is not None
+        pk_col = getattr(model, pk_key)
+
+        raw_fields: Sequence[SearchFieldType] | None = cls.__dict__.get(
+            "searchable_fields", None
+        )
+        if raw_fields is None:
+            cls.searchable_fields = [pk_col]
+        else:
+            if not any(
+                not isinstance(f, tuple) and f.key == pk_key for f in raw_fields
+            ):
+                cls.searchable_fields = [pk_col, *raw_fields]
+
     @classmethod
     def _resolve_load_options(
         cls, load_options: Sequence[ExecutableOption] | None
@@ -869,7 +927,7 @@ class AsyncCrud(Generic[ModelType]):
         facet_fields: Sequence[FacetFieldType] | None = None,
         filter_by: dict[str, Any] | BaseModel | None = None,
         schema: type[BaseModel],
-    ) -> PaginatedResponse[Any]:
+    ) -> OffsetPaginatedResponse[Any]:
         """Get paginated results using offset-based pagination.
 
         Args:
@@ -951,7 +1009,7 @@ class AsyncCrud(Generic[ModelType]):
             session, facet_fields, filters, search_joins
         )
 
-        return PaginatedResponse(
+        return OffsetPaginatedResponse(
             data=items,
             pagination=OffsetPagination(
                 total_count=total_count,
@@ -979,7 +1037,7 @@ class AsyncCrud(Generic[ModelType]):
         facet_fields: Sequence[FacetFieldType] | None = None,
         filter_by: dict[str, Any] | BaseModel | None = None,
         schema: type[BaseModel],
-    ) -> PaginatedResponse[Any]:
+    ) -> CursorPaginatedResponse[Any]:
         """Get paginated results using cursor-based pagination.
 
         Args:
@@ -1018,26 +1076,15 @@ class AsyncCrud(Generic[ModelType]):
         cursor_column: Any = cls.cursor_column
         cursor_col_name: str = cursor_column.key
 
+        direction = _CursorDirection.NEXT
         if cursor is not None:
-            raw_val = _decode_cursor(cursor)
+            raw_val, direction = _decode_cursor(cursor)
             col_type = cursor_column.property.columns[0].type
-            if isinstance(col_type, Integer):
-                cursor_val: Any = int(raw_val)
-            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)
+            cursor_val: Any = _parse_cursor_value(raw_val, col_type)
+            if direction is _CursorDirection.PREV:
+                filters.append(cursor_column < cursor_val)
             else:
-                raise ValueError(
-                    f"Unsupported cursor column type: {type(col_type).__name__!r}. "
-                    "Supported types: Integer, BigInteger, SmallInteger, Uuid, "
-                    "DateTime, Date, Float, Numeric."
-                )
-            filters.append(cursor_column > cursor_val)
+                filters.append(cursor_column > cursor_val)
 
         # Build search filters
         if search:
@@ -1064,12 +1111,15 @@ class AsyncCrud(Generic[ModelType]):
         if resolved := cls._resolve_load_options(load_options):
             q = q.options(*resolved)
 
-        # Cursor column is always the primary sort
-        q = q.order_by(cursor_column)
+        # Cursor column is always the primary sort; reverse direction for prev traversal
+        if direction is _CursorDirection.PREV:
+            q = q.order_by(cursor_column.desc())
+        else:
+            q = q.order_by(cursor_column)
         if order_by is not None:
             q = q.order_by(order_by)
 
-        # Fetch one extra to detect whether a next page exists
+        # Fetch one extra to detect whether another page exists in this direction
         q = q.limit(items_per_page + 1)
         result = await session.execute(q)
         raw_items = cast(list[ModelType], result.unique().scalars().all())
@@ -1077,15 +1127,36 @@ class AsyncCrud(Generic[ModelType]):
         has_more = len(raw_items) > items_per_page
         items_page = raw_items[:items_per_page]
 
-        # next_cursor points past the last item on this page
-        next_cursor: str | None = None
-        if has_more and items_page:
-            next_cursor = _encode_cursor(getattr(items_page[-1], cursor_col_name))
+        # Restore ascending order when traversing backward
+        if direction is _CursorDirection.PREV:
+            items_page = list(reversed(items_page))
 
-        # prev_cursor points to the first item on this page or None when on the first page
+        # next_cursor: points past the last item in ascending order
+        next_cursor: str | None = None
+        if direction is _CursorDirection.NEXT:
+            if has_more and items_page:
+                next_cursor = _encode_cursor(
+                    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=_CursorDirection.NEXT,
+                )
+
+        # prev_cursor: points before the first item in ascending order
         prev_cursor: str | None = None
-        if cursor is not None and items_page:
-            prev_cursor = _encode_cursor(getattr(items_page[0], cursor_col_name))
+        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=_CursorDirection.PREV
+            )
+        elif direction is _CursorDirection.PREV and has_more and items_page:
+            prev_cursor = _encode_cursor(
+                getattr(items_page[0], cursor_col_name), direction=_CursorDirection.PREV
+            )
 
         items: list[Any] = [schema.model_validate(item) for item in items_page]
 
@@ -1093,7 +1164,7 @@ class AsyncCrud(Generic[ModelType]):
             session, facet_fields, filters, search_joins
         )
 
-        return PaginatedResponse(
+        return CursorPaginatedResponse(
             data=items,
             pagination=CursorPagination(
                 next_cursor=next_cursor,
@@ -1104,10 +1175,149 @@ class AsyncCrud(Generic[ModelType]):
             filter_attributes=filter_attributes,
         )
 
+    @overload
+    @classmethod
+    async def paginate(  # pragma: no cover
+        cls: type[Self],
+        session: AsyncSession,
+        *,
+        pagination_type: Literal[PaginationType.OFFSET],
+        filters: list[Any] | None = ...,
+        joins: JoinType | None = ...,
+        outer_join: bool = ...,
+        load_options: Sequence[ExecutableOption] | None = ...,
+        order_by: OrderByClause | None = ...,
+        page: int = ...,
+        cursor: str | None = ...,
+        items_per_page: int = ...,
+        search: str | SearchConfig | None = ...,
+        search_fields: Sequence[SearchFieldType] | None = ...,
+        facet_fields: Sequence[FacetFieldType] | None = ...,
+        filter_by: dict[str, Any] | BaseModel | None = ...,
+        schema: type[BaseModel],
+    ) -> OffsetPaginatedResponse[Any]: ...
+
+    @overload
+    @classmethod
+    async def paginate(  # pragma: no cover
+        cls: type[Self],
+        session: AsyncSession,
+        *,
+        pagination_type: Literal[PaginationType.CURSOR],
+        filters: list[Any] | None = ...,
+        joins: JoinType | None = ...,
+        outer_join: bool = ...,
+        load_options: Sequence[ExecutableOption] | None = ...,
+        order_by: OrderByClause | None = ...,
+        page: int = ...,
+        cursor: str | None = ...,
+        items_per_page: int = ...,
+        search: str | SearchConfig | None = ...,
+        search_fields: Sequence[SearchFieldType] | None = ...,
+        facet_fields: Sequence[FacetFieldType] | None = ...,
+        filter_by: dict[str, Any] | BaseModel | None = ...,
+        schema: type[BaseModel],
+    ) -> CursorPaginatedResponse[Any]: ...
+
+    @classmethod
+    async def paginate(
+        cls: type[Self],
+        session: AsyncSession,
+        *,
+        pagination_type: PaginationType = PaginationType.OFFSET,
+        filters: list[Any] | None = None,
+        joins: JoinType | None = None,
+        outer_join: bool = False,
+        load_options: Sequence[ExecutableOption] | None = None,
+        order_by: OrderByClause | None = None,
+        page: int = 1,
+        cursor: str | None = None,
+        items_per_page: int = 20,
+        search: str | SearchConfig | None = None,
+        search_fields: Sequence[SearchFieldType] | None = None,
+        facet_fields: Sequence[FacetFieldType] | None = None,
+        filter_by: dict[str, Any] | BaseModel | None = None,
+        schema: type[BaseModel],
+    ) -> OffsetPaginatedResponse[Any] | CursorPaginatedResponse[Any]:
+        """Get paginated results using either offset or cursor pagination.
+
+        Args:
+            session: DB async session.
+            pagination_type: Pagination strategy.  Defaults to
+                ``PaginationType.OFFSET``.
+            filters: List of SQLAlchemy filter conditions.
+            joins: List of ``(model, condition)`` tuples for joining related
+                tables.
+            outer_join: Use LEFT OUTER JOIN instead of INNER JOIN.
+            load_options: SQLAlchemy loader options.  Falls back to
+                ``default_load_options`` when not provided.
+            order_by: Column or expression to order results by.
+            page: Page number (1-indexed).  Only used when
+                ``pagination_type`` is ``OFFSET``.
+            cursor: Cursor token from a previous
+                :class:`.CursorPaginatedResponse`.  Only used when
+                ``pagination_type`` is ``CURSOR``.
+            items_per_page: Number of items per page (default 20).
+            search: Search query string or :class:`.SearchConfig` object.
+            search_fields: Fields to search in (overrides class default).
+            facet_fields: Columns to compute distinct values for (overrides
+                class default).
+            filter_by: Dict of ``{column_key: value}`` to filter by declared
+                facet fields.  Keys must match the ``column.key`` of a facet
+                field.  Scalar → equality, list → IN clause.  Raises
+                :exc:`.InvalidFacetFilterError` for unknown keys.
+            schema: Pydantic schema to serialize each item into.
+
+        Returns:
+            :class:`.OffsetPaginatedResponse` when ``pagination_type`` is
+            ``OFFSET``, :class:`.CursorPaginatedResponse` when it is
+            ``CURSOR``.
+        """
+        if items_per_page < 1:
+            raise ValueError(f"items_per_page must be >= 1, got {items_per_page}")
+        match pagination_type:
+            case PaginationType.CURSOR:
+                return await cls.cursor_paginate(
+                    session,
+                    cursor=cursor,
+                    filters=filters,
+                    joins=joins,
+                    outer_join=outer_join,
+                    load_options=load_options,
+                    order_by=order_by,
+                    items_per_page=items_per_page,
+                    search=search,
+                    search_fields=search_fields,
+                    facet_fields=facet_fields,
+                    filter_by=filter_by,
+                    schema=schema,
+                )
+            case PaginationType.OFFSET:
+                if page < 1:
+                    raise ValueError(f"page must be >= 1, got {page}")
+                return await cls.offset_paginate(
+                    session,
+                    filters=filters,
+                    joins=joins,
+                    outer_join=outer_join,
+                    load_options=load_options,
+                    order_by=order_by,
+                    page=page,
+                    items_per_page=items_per_page,
+                    search=search,
+                    search_fields=search_fields,
+                    facet_fields=facet_fields,
+                    filter_by=filter_by,
+                    schema=schema,
+                )
+            case _:
+                raise ValueError(f"Unknown pagination_type: {pagination_type!r}")
+
 
 def CrudFactory(
     model: type[ModelType],
     *,
+    base_class: type[AsyncCrud[Any]] = AsyncCrud,
     searchable_fields: Sequence[SearchFieldType] | None = None,
     facet_fields: Sequence[FacetFieldType] | None = None,
     order_fields: Sequence[QueryableAttribute[Any]] | None = None,
@@ -1119,6 +1329,9 @@ def CrudFactory(
 
     Args:
         model: SQLAlchemy model class
+        base_class: Optional base class to inherit from instead of ``AsyncCrud``.
+            Use this to share custom methods across multiple CRUD classes while
+            still using the factory shorthand.
         searchable_fields: Optional list of searchable fields
         facet_fields: Optional list of columns to compute distinct values for in paginated
             responses. Supports direct columns (``User.status``) and relationship tuples
@@ -1209,28 +1422,27 @@ def CrudFactory(
             joins=[(Post, Post.user_id == User.id)],
             outer_join=True,
         )
+
+        # With a shared custom base class:
+        from typing import Generic, TypeVar
+        from sqlalchemy.orm import DeclarativeBase
+
+        T = TypeVar("T", bound=DeclarativeBase)
+
+        class AuditedCrud(AsyncCrud[T], Generic[T]):
+            @classmethod
+            async def get_active(cls, session):
+                return await cls.get_multi(session, filters=[cls.model.is_active == True])
+
+        UserCrud = CrudFactory(User, base_class=AuditedCrud)
         ```
     """
-    pk_key = model.__mapper__.primary_key[0].key
-    assert pk_key is not None
-    pk_col = getattr(model, pk_key)
-
-    if searchable_fields is None:
-        effective_searchable = [pk_col]
-    else:
-        existing_keys = {f.key for f in searchable_fields if not isinstance(f, tuple)}
-        effective_searchable = (
-            [pk_col, *searchable_fields]
-            if pk_key not in existing_keys
-            else list(searchable_fields)
-        )
-
     cls = type(
         f"Async{model.__name__}Crud",
-        (AsyncCrud,),
+        (base_class,),
         {
             "model": model,
-            "searchable_fields": effective_searchable,
+            "searchable_fields": searchable_fields,
             "facet_fields": facet_fields,
             "order_fields": order_fields,
             "m2m_fields": m2m_fields,

src/fastapi_toolsets/models.py

@@ -3,11 +3,12 @@
 import uuid
 from datetime import datetime
 
-from sqlalchemy import DateTime, Uuid, func, text
+from sqlalchemy import DateTime, Uuid, text
 from sqlalchemy.orm import Mapped, mapped_column
 
 __all__ = [
     "UUIDMixin",
+    "UUIDv7Mixin",
     "CreatedAtMixin",
     "UpdatedAtMixin",
     "TimestampMixin",
@@ -24,12 +25,22 @@ class UUIDMixin:
     )
 
 
+class UUIDv7Mixin:
+    """Mixin that adds a UUIDv7 primary key auto-generated by the database."""
+
+    id: Mapped[uuid.UUID] = mapped_column(
+        Uuid,
+        primary_key=True,
+        server_default=text("uuidv7()"),
+    )
+
+
 class CreatedAtMixin:
     """Mixin that adds a ``created_at`` timestamp column."""
 
     created_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
-        server_default=func.now(),
+        server_default=text("clock_timestamp()"),
     )
 
 
@@ -38,8 +49,8 @@ class UpdatedAtMixin:
 
     updated_at: Mapped[datetime] = mapped_column(
         DateTime(timezone=True),
-        server_default=func.now(),
-        onupdate=func.now(),
+        server_default=text("clock_timestamp()"),
+        onupdate=text("clock_timestamp()"),
     )
 
 

src/fastapi_toolsets/schemas.py

@@ -1,18 +1,21 @@
 """Base Pydantic schemas for API responses."""
 
 from enum import Enum
-from typing import Any, ClassVar, Generic
+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
 
 __all__ = [
     "ApiError",
     "CursorPagination",
+    "CursorPaginatedResponse",
     "ErrorResponse",
     "OffsetPagination",
+    "OffsetPaginatedResponse",
     "PaginatedResponse",
+    "PaginationType",
     "PydanticBase",
     "Response",
     "ResponseStatus",
@@ -123,9 +126,66 @@ class CursorPagination(PydanticBase):
     has_more: bool
 
 
+class PaginationType(str, Enum):
+    """Pagination strategy selector for :meth:`.AsyncCrud.paginate`."""
+
+    OFFSET = "offset"
+    CURSOR = "cursor"
+
+
 class PaginatedResponse(BaseResponse, Generic[DataT]):
-    """Paginated API response for list endpoints."""
+    """Paginated API response for list endpoints.
+
+    Base class and return type for endpoints that support both pagination
+    strategies.  Use :class:`OffsetPaginatedResponse` or
+    :class:`CursorPaginatedResponse` when the strategy is fixed.
+
+    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]
     pagination: OffsetPagination | CursorPagination
+    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.
+
+    The ``pagination_type`` field is always ``"offset"`` and acts as a
+    discriminator, allowing frontend clients to narrow the union type returned
+    by a unified ``paginate()`` endpoint.
+    """
+
+    pagination: OffsetPagination
+    pagination_type: Literal[PaginationType.OFFSET] = PaginationType.OFFSET
+
+
+class CursorPaginatedResponse(PaginatedResponse[DataT]):
+    """Paginated response with typed cursor-based pagination metadata.
+
+    The ``pagination_type`` field is always ``"cursor"`` and acts as a
+    discriminator, allowing frontend clients to narrow the union type returned
+    by a unified ``paginate()`` endpoint.
+    """
+
+    pagination: CursorPagination
+    pagination_type: Literal[PaginationType.CURSOR] = PaginationType.CURSOR

src/fastapi_toolsets/security/__init__.py

@@ -0,0 +1,24 @@
+"""Authentication helpers for FastAPI using Security()."""
+
+from .abc import AuthSource
+from .oauth import (
+    oauth_build_authorization_redirect,
+    oauth_decode_state,
+    oauth_encode_state,
+    oauth_fetch_userinfo,
+    oauth_resolve_provider_urls,
+)
+from .sources import APIKeyHeaderAuth, BearerTokenAuth, CookieAuth, MultiAuth
+
+__all__ = [
+    "APIKeyHeaderAuth",
+    "AuthSource",
+    "BearerTokenAuth",
+    "CookieAuth",
+    "MultiAuth",
+    "oauth_build_authorization_redirect",
+    "oauth_decode_state",
+    "oauth_encode_state",
+    "oauth_fetch_userinfo",
+    "oauth_resolve_provider_urls",
+]

src/fastapi_toolsets/security/abc.py

@@ -0,0 +1,51 @@
+"""Abstract base class for authentication sources."""
+
+import inspect
+from abc import ABC, abstractmethod
+from typing import Any, Callable
+
+from fastapi import Request
+from fastapi.security import SecurityScopes
+
+from fastapi_toolsets.exceptions import UnauthorizedError
+
+
+async def _call_validator(
+    validator: Callable[..., Any], *args: Any, **kwargs: Any
+) -> Any:
+    """Call *validator* with *args* and *kwargs*, awaiting it if it is a coroutine function."""
+    if inspect.iscoroutinefunction(validator):
+        return await validator(*args, **kwargs)
+    return validator(*args, **kwargs)
+
+
+class AuthSource(ABC):
+    """Abstract base class for authentication sources."""
+
+    def __init__(self) -> None:
+        """Set up the default FastAPI dependency signature."""
+        source = self
+
+        async def _call(
+            request: Request,
+            security_scopes: SecurityScopes,  # noqa: ARG001
+        ) -> Any:
+            credential = await source.extract(request)
+            if credential is None:
+                raise UnauthorizedError()
+            return await source.authenticate(credential)
+
+        self._call_fn: Callable[..., Any] = _call
+        self.__signature__ = inspect.signature(_call)
+
+    @abstractmethod
+    async def extract(self, request: Request) -> str | None:
+        """Extract the raw credential from the request without validating."""
+
+    @abstractmethod
+    async def authenticate(self, credential: str) -> Any:
+        """Validate a credential and return the authenticated identity."""
+
+    async def __call__(self, **kwargs: Any) -> Any:
+        """FastAPI dependency dispatch."""
+        return await self._call_fn(**kwargs)

src/fastapi_toolsets/security/oauth.py

@@ -0,0 +1,140 @@
+"""OAuth 2.0 / OIDC helper utilities."""
+
+import base64
+from typing import Any
+from urllib.parse import urlencode
+
+import httpx
+from fastapi.responses import RedirectResponse
+
+_discovery_cache: dict[str, dict] = {}
+
+
+async def oauth_resolve_provider_urls(
+    discovery_url: str,
+) -> tuple[str, str, str | None]:
+    """Fetch the OIDC discovery document and return endpoint URLs.
+
+    Args:
+        discovery_url: URL of the provider's ``/.well-known/openid-configuration``.
+
+    Returns:
+        A ``(authorization_url, token_url, userinfo_url)`` tuple.
+        *userinfo_url* is ``None`` when the provider does not advertise one.
+    """
+    if discovery_url not in _discovery_cache:
+        async with httpx.AsyncClient() as client:
+            resp = await client.get(discovery_url)
+            resp.raise_for_status()
+            _discovery_cache[discovery_url] = resp.json()
+    cfg = _discovery_cache[discovery_url]
+    return (
+        cfg["authorization_endpoint"],
+        cfg["token_endpoint"],
+        cfg.get("userinfo_endpoint"),
+    )
+
+
+async def oauth_fetch_userinfo(
+    *,
+    token_url: str,
+    userinfo_url: str,
+    code: str,
+    client_id: str,
+    client_secret: str,
+    redirect_uri: str,
+) -> dict[str, Any]:
+    """Exchange an authorization code for tokens and return the userinfo payload.
+
+    Performs the two-step OAuth 2.0 / OIDC token exchange:
+
+    1. POSTs the authorization *code* to *token_url* to obtain an access token.
+    2. GETs *userinfo_url* using that access token as a Bearer credential.
+
+    Args:
+        token_url: Provider's token endpoint.
+        userinfo_url: Provider's userinfo endpoint.
+        code: Authorization code received from the provider's callback.
+        client_id: OAuth application client ID.
+        client_secret: OAuth application client secret.
+        redirect_uri: Redirect URI that was used in the authorization request.
+
+    Returns:
+        The JSON payload returned by the userinfo endpoint as a plain ``dict``.
+    """
+    async with httpx.AsyncClient() as client:
+        token_resp = await client.post(
+            token_url,
+            data={
+                "grant_type": "authorization_code",
+                "code": code,
+                "client_id": client_id,
+                "client_secret": client_secret,
+                "redirect_uri": redirect_uri,
+            },
+            headers={"Accept": "application/json"},
+        )
+        token_resp.raise_for_status()
+        access_token = token_resp.json()["access_token"]
+
+        userinfo_resp = await client.get(
+            userinfo_url,
+            headers={"Authorization": f"Bearer {access_token}"},
+        )
+        userinfo_resp.raise_for_status()
+        return userinfo_resp.json()
+
+
+def oauth_build_authorization_redirect(
+    authorization_url: str,
+    *,
+    client_id: str,
+    scopes: str,
+    redirect_uri: str,
+    destination: str,
+) -> RedirectResponse:
+    """Return an OAuth 2.0 authorization ``RedirectResponse``.
+
+    Args:
+        authorization_url: Provider's authorization endpoint.
+        client_id: OAuth application client ID.
+        scopes: Space-separated list of requested scopes.
+        redirect_uri: URI the provider should redirect back to after authorization.
+        destination: URL the user should be sent to after the full OAuth flow
+            completes (encoded as ``state``).
+
+    Returns:
+        A :class:`~fastapi.responses.RedirectResponse` to the provider's
+        authorization page.
+    """
+    params = urlencode(
+        {
+            "client_id": client_id,
+            "response_type": "code",
+            "scope": scopes,
+            "redirect_uri": redirect_uri,
+            "state": oauth_encode_state(destination),
+        }
+    )
+    return RedirectResponse(f"{authorization_url}?{params}")
+
+
+def oauth_encode_state(url: str) -> str:
+    """Base64url-encode a URL to embed as an OAuth ``state`` parameter."""
+    return base64.urlsafe_b64encode(url.encode()).decode()
+
+
+def oauth_decode_state(state: str | None, *, fallback: str) -> str:
+    """Decode a base64url OAuth ``state`` parameter.
+
+    Handles missing padding (some providers strip ``=``).
+    Returns *fallback* if *state* is absent, the literal string ``"null"``,
+    or cannot be decoded.
+    """
+    if not state or state == "null":
+        return fallback
+    try:
+        padded = state + "=" * (4 - len(state) % 4)
+        return base64.urlsafe_b64decode(padded).decode()
+    except Exception:
+        return fallback

src/fastapi_toolsets/security/sources/__init__.py

@@ -0,0 +1,8 @@
+"""Built-in authentication source implementations."""
+
+from .header import APIKeyHeaderAuth
+from .bearer import BearerTokenAuth
+from .cookie import CookieAuth
+from .multi import MultiAuth
+
+__all__ = ["APIKeyHeaderAuth", "BearerTokenAuth", "CookieAuth", "MultiAuth"]

src/fastapi_toolsets/security/sources/bearer.py

@@ -0,0 +1,122 @@
+"""Bearer token authentication source."""
+
+import inspect
+import secrets
+from typing import Annotated, Any, Callable
+
+from fastapi import Depends
+from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer, SecurityScopes
+
+from fastapi_toolsets.exceptions import UnauthorizedError
+
+from ..abc import AuthSource, _call_validator
+
+
+class BearerTokenAuth(AuthSource):
+    """Bearer token authentication source.
+
+    Wraps :class:`fastapi.security.HTTPBearer` for OpenAPI documentation.
+    The validator is called as ``await validator(credential, **kwargs)``
+    where ``kwargs`` are the extra keyword arguments provided at instantiation.
+
+    Args:
+        validator: Sync or async callable that receives the credential and any
+            extra keyword arguments, and returns the authenticated identity
+            (e.g. a ``User`` model). Should raise
+            :class:`~fastapi_toolsets.exceptions.UnauthorizedError` on failure.
+        prefix: Optional token prefix (e.g. ``"user_"``). If set, only tokens
+            whose value starts with this prefix are matched. The prefix is
+            **kept** in the value passed to the validator — store and compare
+            tokens with their prefix included. Use :meth:`generate_token` to
+            create correctly-prefixed tokens. This enables multiple
+            ``BearerTokenAuth`` instances in the same app (e.g. ``"user_"``
+            for user tokens, ``"org_"`` for org tokens).
+        **kwargs: Extra keyword arguments forwarded to the validator on every
+            call (e.g. ``role=Role.ADMIN``).
+    """
+
+    def __init__(
+        self,
+        validator: Callable[..., Any],
+        *,
+        prefix: str | None = None,
+        **kwargs: Any,
+    ) -> None:
+        self._validator = validator
+        self._prefix = prefix
+        self._kwargs = kwargs
+        self._scheme = HTTPBearer(auto_error=False)
+
+        _scheme = self._scheme
+        _validator = validator
+        _kwargs = kwargs
+        _prefix = prefix
+
+        async def _call(
+            security_scopes: SecurityScopes,  # noqa: ARG001
+            credentials: Annotated[
+                HTTPAuthorizationCredentials | None, Depends(_scheme)
+            ] = None,
+        ) -> Any:
+            if credentials is None:
+                raise UnauthorizedError()
+            token = credentials.credentials
+            if _prefix is not None and not token.startswith(_prefix):
+                raise UnauthorizedError()
+            return await _call_validator(_validator, token, **_kwargs)
+
+        self._call_fn = _call
+        self.__signature__ = inspect.signature(_call)
+
+    async def extract(self, request: Any) -> str | None:
+        """Extract the raw credential from the request without validating.
+
+        Returns ``None`` if no ``Authorization: Bearer`` header is present,
+        the token is empty, or the token does not match the configured prefix.
+        The prefix is included in the returned value.
+        """
+        auth = request.headers.get("Authorization", "")
+        if not auth.startswith("Bearer "):
+            return None
+        token = auth[7:]
+        if not token:
+            return None
+        if self._prefix is not None and not token.startswith(self._prefix):
+            return None
+        return token
+
+    async def authenticate(self, credential: str) -> Any:
+        """Validate a credential and return the identity.
+
+        Calls ``await validator(credential, **kwargs)`` where ``kwargs`` are
+        the extra keyword arguments provided at instantiation.
+        """
+        return await _call_validator(self._validator, credential, **self._kwargs)
+
+    def require(self, **kwargs: Any) -> "BearerTokenAuth":
+        """Return a new instance with additional (or overriding) validator kwargs."""
+        return BearerTokenAuth(
+            self._validator,
+            prefix=self._prefix,
+            **{**self._kwargs, **kwargs},
+        )
+
+    def generate_token(self, nbytes: int = 32) -> str:
+        """Generate a secure random token for this auth source.
+
+        Returns a URL-safe random token. If a prefix is configured it is
+        prepended — the returned value is what you store in your database
+        and return to the client as-is.
+
+        Args:
+            nbytes: Number of random bytes before base64 encoding. The
+                resulting string is ``ceil(nbytes * 4 / 3)`` characters
+                (43 chars for the default 32 bytes). Defaults to 32.
+
+        Returns:
+            A ready-to-use token string (e.g. ``"user_Xk3..."``).
+        """
+        token = secrets.token_urlsafe(nbytes)
+        if self._prefix is not None:
+            return f"{self._prefix}{token}"
+        return token

src/fastapi_toolsets/security/sources/cookie.py

@@ -0,0 +1,142 @@
+"""Cookie-based authentication source."""
+
+import base64
+import hashlib
+import hmac
+import inspect
+import json
+import time
+from typing import Annotated, Any, Callable
+
+from fastapi import Depends, Request, Response
+from fastapi.security import APIKeyCookie, SecurityScopes
+
+from fastapi_toolsets.exceptions import UnauthorizedError
+
+from ..abc import AuthSource, _call_validator
+
+
+class CookieAuth(AuthSource):
+    """Cookie-based authentication source.
+
+    Wraps :class:`fastapi.security.APIKeyCookie` for OpenAPI documentation.
+    Optionally signs the cookie with HMAC-SHA256 to provide stateless, tamper-
+    proof sessions without any database entry.
+
+    Args:
+        name: Cookie name.
+        validator: Sync or async callable that receives the cookie value
+            (plain, after signature verification when ``secret_key`` is set)
+            and any extra keyword arguments, and returns the authenticated
+            identity.
+        secret_key: When provided, the cookie is HMAC-SHA256 signed.
+            :meth:`set_cookie` embeds an expiry and signs the payload;
+            :meth:`extract` verifies the signature and expiry before handing
+            the plain value to the validator.  When ``None`` (default), the raw
+            cookie value is passed to the validator as-is.
+        ttl: Cookie lifetime in seconds (default 24 h).  Only used when
+            ``secret_key`` is set.
+        **kwargs: Extra keyword arguments forwarded to the validator on every
+            call (e.g. ``role=Role.ADMIN``).
+    """
+
+    def __init__(
+        self,
+        name: str,
+        validator: Callable[..., Any],
+        *,
+        secret_key: str | None = None,
+        ttl: int = 86400,
+        **kwargs: Any,
+    ) -> None:
+        self._name = name
+        self._validator = validator
+        self._secret_key = secret_key
+        self._ttl = ttl
+        self._kwargs = kwargs
+        self._scheme = APIKeyCookie(name=name, auto_error=False)
+
+        _scheme = self._scheme
+        _self = self
+        _kwargs = kwargs
+
+        async def _call(
+            security_scopes: SecurityScopes,  # noqa: ARG001
+            value: Annotated[str | None, Depends(_scheme)] = None,
+        ) -> Any:
+            if value is None:
+                raise UnauthorizedError()
+            plain = _self._verify(value)
+            return await _call_validator(_self._validator, plain, **_kwargs)
+
+        self._call_fn = _call
+        self.__signature__ = inspect.signature(_call)
+
+    def _hmac(self, data: str) -> str:
+        assert self._secret_key is not None
+        return hmac.new(
+            self._secret_key.encode(), data.encode(), hashlib.sha256
+        ).hexdigest()
+
+    def _sign(self, value: str) -> str:
+        data = base64.urlsafe_b64encode(
+            json.dumps({"v": value, "exp": int(time.time()) + self._ttl}).encode()
+        ).decode()
+        return f"{data}.{self._hmac(data)}"
+
+    def _verify(self, cookie_value: str) -> str:
+        """Return the plain value, verifying HMAC + expiry when signed."""
+        if not self._secret_key:
+            return cookie_value
+
+        try:
+            data, sig = cookie_value.rsplit(".", 1)
+        except ValueError:
+            raise UnauthorizedError()
+
+        if not hmac.compare_digest(self._hmac(data), sig):
+            raise UnauthorizedError()
+
+        try:
+            payload = json.loads(base64.urlsafe_b64decode(data))
+            value: str = payload["v"]
+            exp: int = payload["exp"]
+        except Exception:
+            raise UnauthorizedError()
+
+        if exp < int(time.time()):
+            raise UnauthorizedError()
+
+        return value
+
+    async def extract(self, request: Request) -> str | None:
+        return request.cookies.get(self._name)
+
+    async def authenticate(self, credential: str) -> Any:
+        plain = self._verify(credential)
+        return await _call_validator(self._validator, plain, **self._kwargs)
+
+    def require(self, **kwargs: Any) -> "CookieAuth":
+        """Return a new instance with additional (or overriding) validator kwargs."""
+        return CookieAuth(
+            self._name,
+            self._validator,
+            secret_key=self._secret_key,
+            ttl=self._ttl,
+            **{**self._kwargs, **kwargs},
+        )
+
+    def set_cookie(self, response: Response, value: str) -> None:
+        """Attach the cookie to *response*, signing it when ``secret_key`` is set."""
+        cookie_value = self._sign(value) if self._secret_key else value
+        response.set_cookie(
+            self._name,
+            cookie_value,
+            httponly=True,
+            samesite="lax",
+            max_age=self._ttl,
+        )
+
+    def delete_cookie(self, response: Response) -> None:
+        """Clear the session cookie (logout)."""
+        response.delete_cookie(self._name, httponly=True, samesite="lax")

src/fastapi_toolsets/security/sources/header.py

@@ -0,0 +1,71 @@
+"""API key header authentication source."""
+
+import inspect
+from typing import Annotated, Any, Callable
+
+from fastapi import Depends, Request
+from fastapi.security import APIKeyHeader, SecurityScopes
+
+from fastapi_toolsets.exceptions import UnauthorizedError
+
+from ..abc import AuthSource, _call_validator
+
+
+class APIKeyHeaderAuth(AuthSource):
+    """API key header authentication source.
+
+    Wraps :class:`fastapi.security.APIKeyHeader` for OpenAPI documentation.
+    The validator is called as ``await validator(api_key, **kwargs)``
+    where ``kwargs`` are the extra keyword arguments provided at instantiation.
+
+    Args:
+        name: HTTP header name that carries the API key (e.g. ``"X-API-Key"``).
+        validator: Sync or async callable that receives the API key and any
+            extra keyword arguments, and returns the authenticated identity.
+            Should raise :class:`~fastapi_toolsets.exceptions.UnauthorizedError`
+            on failure.
+        **kwargs: Extra keyword arguments forwarded to the validator on every
+            call (e.g. ``role=Role.ADMIN``).
+    """
+
+    def __init__(
+        self,
+        name: str,
+        validator: Callable[..., Any],
+        **kwargs: Any,
+    ) -> None:
+        self._name = name
+        self._validator = validator
+        self._kwargs = kwargs
+        self._scheme = APIKeyHeader(name=name, auto_error=False)
+
+        _scheme = self._scheme
+        _validator = validator
+        _kwargs = kwargs
+
+        async def _call(
+            security_scopes: SecurityScopes,  # noqa: ARG001
+            api_key: Annotated[str | None, Depends(_scheme)] = None,
+        ) -> Any:
+            if api_key is None:
+                raise UnauthorizedError()
+            return await _call_validator(_validator, api_key, **_kwargs)
+
+        self._call_fn = _call
+        self.__signature__ = inspect.signature(_call)
+
+    async def extract(self, request: Request) -> str | None:
+        """Extract the API key from the configured header."""
+        return request.headers.get(self._name) or None
+
+    async def authenticate(self, credential: str) -> Any:
+        """Validate a credential and return the identity."""
+        return await _call_validator(self._validator, credential, **self._kwargs)
+
+    def require(self, **kwargs: Any) -> "APIKeyHeaderAuth":
+        """Return a new instance with additional (or overriding) validator kwargs."""
+        return APIKeyHeaderAuth(
+            self._name,
+            self._validator,
+            **{**self._kwargs, **kwargs},
+        )

src/fastapi_toolsets/security/sources/multi.py

@@ -0,0 +1,121 @@
+"""MultiAuth: combine multiple authentication sources into a single callable."""
+
+import inspect
+from typing import Any, cast
+
+from fastapi import Request
+from fastapi.security import SecurityScopes
+
+from fastapi_toolsets.exceptions import UnauthorizedError
+
+from ..abc import AuthSource
+
+
+class MultiAuth:
+    """Combine multiple authentication sources into a single callable.
+
+    Sources are tried in order; the first one whose
+    :meth:`~AuthSource.extract` returns a non-``None`` credential wins.
+    Its :meth:`~AuthSource.authenticate` is called and the result returned.
+
+    If a credential is found but the validator raises, the exception propagates
+    immediately — the remaining sources are **not** tried. This prevents
+    silent fallthrough on invalid credentials.
+
+    If no source provides a credential,
+    :class:`~fastapi_toolsets.exceptions.UnauthorizedError` is raised.
+
+    The :meth:`~AuthSource.extract` method of each source performs only
+    string matching (no I/O), so prefix-based dispatch is essentially free.
+
+    Any :class:`~AuthSource` subclass — including user-defined ones — can be
+    passed as a source.
+
+    Args:
+        *sources: Auth source instances to try in order.
+
+    Example::
+
+        user_bearer = BearerTokenAuth(verify_user, prefix="user_")
+        org_bearer  = BearerTokenAuth(verify_org,  prefix="org_")
+        cookie      = CookieAuth("session", verify_session)
+
+        multi = MultiAuth(user_bearer, org_bearer, cookie)
+
+        @app.get("/data")
+        async def data_route(user = Security(multi)):
+            return user
+
+        # Apply a shared requirement to all sources at once
+        @app.get("/admin")
+        async def admin_route(user = Security(multi.require(role=Role.ADMIN))):
+            return user
+    """
+
+    def __init__(self, *sources: AuthSource) -> None:
+        self._sources = sources
+
+        _sources = sources
+
+        async def _call(
+            request: Request,
+            security_scopes: SecurityScopes,  # noqa: ARG001
+            **kwargs: Any,  # noqa: ARG001  — absorbs scheme values injected by FastAPI
+        ) -> Any:
+            for source in _sources:
+                credential = await source.extract(request)
+                if credential is not None:
+                    return await source.authenticate(credential)
+            raise UnauthorizedError()
+
+        self._call_fn = _call
+
+        # Build a merged signature that includes the security-scheme Depends()
+        # parameters from every source so FastAPI registers them in OpenAPI docs.
+        seen: set[str] = {"request", "security_scopes"}
+        merged: list[inspect.Parameter] = [
+            inspect.Parameter(
+                "request",
+                inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                annotation=Request,
+            ),
+            inspect.Parameter(
+                "security_scopes",
+                inspect.Parameter.POSITIONAL_OR_KEYWORD,
+                annotation=SecurityScopes,
+            ),
+        ]
+        for i, source in enumerate(sources):
+            for name, param in inspect.signature(source).parameters.items():
+                if name in seen:
+                    continue
+                merged.append(param.replace(name=f"_s{i}_{name}"))
+                seen.add(name)
+        self.__signature__ = inspect.Signature(merged, return_annotation=Any)
+
+    async def __call__(self, **kwargs: Any) -> Any:
+        return await self._call_fn(**kwargs)
+
+    def require(self, **kwargs: Any) -> "MultiAuth":
+        """Return a new :class:`MultiAuth` with kwargs forwarded to each source.
+
+        Calls ``.require(**kwargs)`` on every source that supports it. Sources
+        that do not implement ``.require()`` (e.g. custom :class:`~AuthSource`
+        subclasses) are passed through unchanged.
+
+        New kwargs are merged over each source's existing kwargs — new values
+        win on conflict::
+
+            multi = MultiAuth(bearer, cookie)
+
+            @app.get("/admin")
+            async def admin(user = Security(multi.require(role=Role.ADMIN))):
+                return user
+        """
+        new_sources = tuple(
+            cast(Any, source).require(**kwargs)
+            if hasattr(source, "require")
+            else source
+            for source in self._sources
+        )
+        return MultiAuth(*new_sources)

tests/test_crud.py

@@ -6,8 +6,8 @@ import pytest
 from sqlalchemy.ext.asyncio import AsyncSession
 from sqlalchemy.orm import selectinload
 
-from fastapi_toolsets.crud import CrudFactory
-from fastapi_toolsets.crud.factory import AsyncCrud
+from fastapi_toolsets.crud import CrudFactory, PaginationType
+from fastapi_toolsets.crud.factory import AsyncCrud, _CursorDirection
 from fastapi_toolsets.exceptions import NotFoundError
 
 from .conftest import (
@@ -86,6 +86,101 @@ class TestCrudFactory:
         assert crud_with.default_load_options == options
         assert crud_without.default_load_options is None
 
+    def test_base_class_custom_methods_inherited(self):
+        """CrudFactory with base_class inherits custom methods from that base."""
+        from typing import Generic, TypeVar
+
+        from sqlalchemy.orm import DeclarativeBase
+
+        T = TypeVar("T", bound=DeclarativeBase)
+
+        class CustomBase(AsyncCrud[T], Generic[T]):
+            @classmethod
+            def custom_method(cls) -> str:
+                return f"custom:{cls.model.__name__}"
+
+        UserCrudCustom = CrudFactory(User, base_class=CustomBase)
+        PostCrudCustom = CrudFactory(Post, base_class=CustomBase)
+
+        assert issubclass(UserCrudCustom, CustomBase)
+        assert issubclass(PostCrudCustom, CustomBase)
+        assert UserCrudCustom.custom_method() == "custom:User"
+        assert PostCrudCustom.custom_method() == "custom:Post"
+
+    def test_base_class_pk_injected(self):
+        """PK is still injected when using a custom base_class."""
+        from typing import Generic, TypeVar
+
+        from sqlalchemy.orm import DeclarativeBase
+
+        T = TypeVar("T", bound=DeclarativeBase)
+
+        class CustomBase(AsyncCrud[T], Generic[T]):
+            pass
+
+        crud = CrudFactory(User, base_class=CustomBase)
+        assert crud.searchable_fields is not None
+        assert User.id in crud.searchable_fields
+
+
+class TestAsyncCrudSubclass:
+    """Tests for direct AsyncCrud subclassing (alternative to CrudFactory)."""
+
+    def test_subclass_with_model_only(self):
+        """Subclassing with just model auto-injects PK into searchable_fields."""
+
+        class UserCrudDirect(AsyncCrud[User]):
+            model = User
+
+        assert UserCrudDirect.searchable_fields == [User.id]
+
+    def test_subclass_with_explicit_fields_prepends_pk(self):
+        """Subclassing with searchable_fields prepends PK automatically."""
+
+        class UserCrudDirect(AsyncCrud[User]):
+            model = User
+            searchable_fields = [User.username]
+
+        assert UserCrudDirect.searchable_fields == [User.id, User.username]
+
+    def test_subclass_with_pk_already_in_fields(self):
+        """PK is not duplicated when already in searchable_fields."""
+
+        class UserCrudDirect(AsyncCrud[User]):
+            model = User
+            searchable_fields = [User.id, User.username]
+
+        assert UserCrudDirect.searchable_fields == [User.id, User.username]
+
+    def test_subclass_has_default_class_vars(self):
+        """Other ClassVars are None by default on a direct subclass."""
+
+        class UserCrudDirect(AsyncCrud[User]):
+            model = User
+
+        assert UserCrudDirect.facet_fields is None
+        assert UserCrudDirect.default_load_options is None
+        assert UserCrudDirect.cursor_column is None
+
+    def test_subclass_with_load_options(self):
+        """Direct subclass can declare default_load_options."""
+        opts = [selectinload(User.role)]
+
+        class UserCrudDirect(AsyncCrud[User]):
+            model = User
+            default_load_options = opts
+
+        assert UserCrudDirect.default_load_options is opts
+
+    def test_abstract_base_without_model_not_processed(self):
+        """Intermediate abstract class without model is not processed."""
+
+        class AbstractCrud(AsyncCrud[User]):
+            pass
+
+        # Should not raise, and searchable_fields inherits base default (None)
+        assert AbstractCrud.searchable_fields is None
+
 
 class TestResolveLoadOptions:
     """Tests for _resolve_load_options logic."""
@@ -1874,11 +1969,8 @@ class TestCursorPaginatePrevCursor:
         assert page2.pagination.prev_cursor is not None
 
     @pytest.mark.anyio
-    async def test_prev_cursor_points_to_first_item(self, db_session: AsyncSession):
-        """prev_cursor encodes the value of the first item on the current page."""
-        import base64
-        import json
-
+    async def test_prev_cursor_navigates_back(self, db_session: AsyncSession):
+        """prev_cursor on page 2 navigates back to the same items as page 1."""
         for i in range(10):
             await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
 
@@ -1897,12 +1989,83 @@ class TestCursorPaginatePrevCursor:
         assert isinstance(page2.pagination, CursorPagination)
         assert page2.pagination.prev_cursor is not None
 
-        # Decode prev_cursor and compare to first item's id
-        decoded = json.loads(
-            base64.b64decode(page2.pagination.prev_cursor.encode()).decode()
+        # Using prev_cursor should return the same items as page 1
+        back_to_page1 = await RoleCursorCrud.cursor_paginate(
+            db_session,
+            cursor=page2.pagination.prev_cursor,
+            items_per_page=5,
+            schema=RoleRead,
         )
-        first_item_id = str(page2.data[0].id)
-        assert decoded == first_item_id
+        assert isinstance(back_to_page1.pagination, CursorPagination)
+        assert [r.id for r in back_to_page1.data] == [r.id for r in page1.data]
+        assert back_to_page1.pagination.prev_cursor is None
+
+    @pytest.mark.anyio
+    async def test_prev_cursor_empty_result_when_no_items_before(
+        self, db_session: AsyncSession
+    ):
+        """Going backward past the first item returns an empty page."""
+        from fastapi_toolsets.crud.factory import _encode_cursor
+        from fastapi_toolsets.schemas import CursorPagination
+
+        await IntRoleCursorCrud.create(db_session, IntRoleCreate(name="role00"))
+
+        page1 = await IntRoleCursorCrud.cursor_paginate(
+            db_session, items_per_page=5, schema=IntRoleRead
+        )
+        assert isinstance(page1.pagination, CursorPagination)
+
+        # Manually craft a backward cursor before any existing id
+        before_all = _encode_cursor(0, direction=_CursorDirection.PREV)
+        empty = await IntRoleCursorCrud.cursor_paginate(
+            db_session, cursor=before_all, items_per_page=5, schema=IntRoleRead
+        )
+
+        assert isinstance(empty.pagination, CursorPagination)
+        assert empty.data == []
+        assert empty.pagination.next_cursor is None
+        assert empty.pagination.prev_cursor is None
+
+    @pytest.mark.anyio
+    async def test_prev_cursor_set_when_more_pages_behind(
+        self, db_session: AsyncSession
+    ):
+        """Going backward on page 2 (of 3) still exposes a prev_cursor for page 1."""
+        for i in range(9):
+            await RoleCrud.create(db_session, RoleCreate(name=f"role{i:02d}"))
+
+        from fastapi_toolsets.schemas import CursorPagination
+
+        page1 = await RoleCursorCrud.cursor_paginate(
+            db_session, items_per_page=3, schema=RoleRead
+        )
+        assert isinstance(page1.pagination, CursorPagination)
+        page2 = await RoleCursorCrud.cursor_paginate(
+            db_session,
+            cursor=page1.pagination.next_cursor,
+            items_per_page=3,
+            schema=RoleRead,
+        )
+        assert isinstance(page2.pagination, CursorPagination)
+        page3 = await RoleCursorCrud.cursor_paginate(
+            db_session,
+            cursor=page2.pagination.next_cursor,
+            items_per_page=3,
+            schema=RoleRead,
+        )
+        assert isinstance(page3.pagination, CursorPagination)
+        assert page3.pagination.prev_cursor is not None
+
+        # Going back to page 2 should still have a prev_cursor pointing at page 1
+        back_to_page2 = await RoleCursorCrud.cursor_paginate(
+            db_session,
+            cursor=page3.pagination.prev_cursor,
+            items_per_page=3,
+            schema=RoleRead,
+        )
+        assert isinstance(back_to_page2.pagination, CursorPagination)
+        assert [r.id for r in back_to_page2.data] == [r.id for r in page2.data]
+        assert back_to_page2.pagination.prev_cursor is not None
 
 
 class TestCursorPaginateWithSearch:
@@ -2289,3 +2452,72 @@ class TestCursorPaginateColumnTypes:
         page1_ids = {p.id for p in page1.data}
         page2_ids = {p.id for p in page2.data}
         assert page1_ids.isdisjoint(page2_ids)
+
+
+class TestPaginate:
+    """Tests for the unified paginate() method."""
+
+    @pytest.mark.anyio
+    async def test_offset_pagination(self, db_session: AsyncSession):
+        """paginate() with OFFSET returns OffsetPaginatedResponse."""
+        from fastapi_toolsets.schemas import OffsetPagination
+
+        await RoleCrud.create(db_session, RoleCreate(name="admin"))
+        await RoleCrud.create(db_session, RoleCreate(name="user"))
+
+        result = await RoleCrud.paginate(
+            db_session,
+            pagination_type=PaginationType.OFFSET,
+            schema=RoleRead,
+        )
+
+        assert isinstance(result.pagination, OffsetPagination)
+        assert result.pagination_type == PaginationType.OFFSET
+
+    @pytest.mark.anyio
+    async def test_cursor_pagination(self, db_session: AsyncSession):
+        """paginate() with CURSOR returns CursorPaginatedResponse."""
+        from fastapi_toolsets.schemas import CursorPagination
+
+        await RoleCursorCrud.create(db_session, RoleCreate(name="admin"))
+
+        result = await RoleCursorCrud.paginate(
+            db_session,
+            pagination_type=PaginationType.CURSOR,
+            schema=RoleRead,
+        )
+
+        assert isinstance(result.pagination, CursorPagination)
+        assert result.pagination_type == PaginationType.CURSOR
+
+    @pytest.mark.anyio
+    async def test_invalid_items_per_page_raises(self, db_session: AsyncSession):
+        """paginate() raises ValueError when items_per_page < 1."""
+        with pytest.raises(ValueError, match="items_per_page"):
+            await RoleCrud.paginate(
+                db_session,
+                pagination_type=PaginationType.OFFSET,
+                items_per_page=0,
+                schema=RoleRead,
+            )
+
+    @pytest.mark.anyio
+    async def test_invalid_page_raises(self, db_session: AsyncSession):
+        """paginate() raises ValueError when page < 1 for offset pagination."""
+        with pytest.raises(ValueError, match="page"):
+            await RoleCrud.paginate(
+                db_session,
+                pagination_type=PaginationType.OFFSET,
+                page=0,
+                schema=RoleRead,
+            )
+
+    @pytest.mark.anyio
+    async def test_unknown_pagination_type_raises(self, db_session: AsyncSession):
+        """paginate() raises ValueError for unknown pagination_type."""
+        with pytest.raises(ValueError, match="Unknown pagination_type"):
+            await RoleCrud.paginate(
+                db_session,
+                pagination_type="unknown",
+                schema=RoleRead,
+            )  # type: ignore[no-matching-overload]

tests/test_example_pagination_search.py

@@ -393,3 +393,105 @@ class TestCursorSorting:
         body = resp.json()
         assert body["error_code"] == "SORT-422"
         assert body["status"] == "FAIL"
+
+
+class TestPaginateUnified:
+    """Tests for the unified GET /articles/ endpoint using paginate()."""
+
+    @pytest.mark.anyio
+    async def test_defaults_to_offset_pagination(
+        self, client: AsyncClient, ex_db_session
+    ):
+        """Without pagination_type, defaults to offset pagination."""
+        await seed(ex_db_session)
+
+        resp = await client.get("/articles/")
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "offset"
+        assert "total_count" in body["pagination"]
+        assert body["pagination"]["total_count"] == 3
+
+    @pytest.mark.anyio
+    async def test_explicit_offset_pagination(self, client: AsyncClient, ex_db_session):
+        """pagination_type=offset returns OffsetPagination metadata."""
+        await seed(ex_db_session)
+
+        resp = await client.get(
+            "/articles/?pagination_type=offset&page=1&items_per_page=2"
+        )
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "offset"
+        assert body["pagination"]["total_count"] == 3
+        assert body["pagination"]["page"] == 1
+        assert body["pagination"]["has_more"] is True
+        assert len(body["data"]) == 2
+
+    @pytest.mark.anyio
+    async def test_cursor_pagination_type(self, client: AsyncClient, ex_db_session):
+        """pagination_type=cursor returns CursorPagination metadata."""
+        await seed(ex_db_session)
+
+        resp = await client.get("/articles/?pagination_type=cursor&items_per_page=2")
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "cursor"
+        assert "next_cursor" in body["pagination"]
+        assert "total_count" not in body["pagination"]
+        assert body["pagination"]["has_more"] is True
+        assert len(body["data"]) == 2
+
+    @pytest.mark.anyio
+    async def test_cursor_pagination_navigate_pages(
+        self, client: AsyncClient, ex_db_session
+    ):
+        """Cursor from first page can be used to fetch the next page."""
+        await seed(ex_db_session)
+
+        first = await client.get("/articles/?pagination_type=cursor&items_per_page=2")
+        assert first.status_code == 200
+        first_body = first.json()
+        next_cursor = first_body["pagination"]["next_cursor"]
+        assert next_cursor is not None
+
+        second = await client.get(
+            f"/articles/?pagination_type=cursor&items_per_page=2&cursor={next_cursor}"
+        )
+        assert second.status_code == 200
+        second_body = second.json()
+        assert second_body["pagination_type"] == "cursor"
+        assert second_body["pagination"]["has_more"] is False
+        assert len(second_body["data"]) == 1
+
+    @pytest.mark.anyio
+    async def test_cursor_pagination_with_search(
+        self, client: AsyncClient, ex_db_session
+    ):
+        """paginate() with cursor type respects search parameter."""
+        await seed(ex_db_session)
+
+        resp = await client.get("/articles/?pagination_type=cursor&search=fastapi")
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "cursor"
+        assert len(body["data"]) == 1
+        assert body["data"][0]["title"] == "FastAPI tips"
+
+    @pytest.mark.anyio
+    async def test_offset_pagination_with_filter(
+        self, client: AsyncClient, ex_db_session
+    ):
+        """paginate() with offset type respects filter_by parameter."""
+        await seed(ex_db_session)
+
+        resp = await client.get("/articles/?pagination_type=offset&status=published")
+
+        assert resp.status_code == 200
+        body = resp.json()
+        assert body["pagination_type"] == "offset"
+        assert body["pagination"]["total_count"] == 2

tests/test_models.py

@@ -11,6 +11,7 @@ from fastapi_toolsets.models import (
     CreatedAtMixin,
     TimestampMixin,
     UUIDMixin,
+    UUIDv7Mixin,
     UpdatedAtMixin,
 )
 
@@ -48,6 +49,12 @@ class TimestampModel(MixinBase, TimestampMixin):
     name: Mapped[str] = mapped_column(String(50))
 
 
+class UUIDv7Model(MixinBase, UUIDv7Mixin):
+    __tablename__ = "mixin_uuidv7_models"
+
+    name: Mapped[str] = mapped_column(String(50))
+
+
 class FullMixinModel(MixinBase, UUIDMixin, UpdatedAtMixin):
     __tablename__ = "mixin_full_models"
 
@@ -233,6 +240,50 @@ class TestTimestampMixin:
         assert "updated_at" in col_names
 
 
+class TestUUIDv7Mixin:
+    @pytest.mark.anyio
+    async def test_uuid7_generated_by_db(self, mixin_session):
+        """UUIDv7 is generated server-side and populated after flush."""
+        obj = UUIDv7Model(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_uuid7_is_primary_key(self):
+        """UUIDv7Mixin adds id as primary key column."""
+        pk_cols = [c.name for c in UUIDv7Model.__table__.primary_key]
+        assert pk_cols == ["id"]
+
+    @pytest.mark.anyio
+    async def test_each_row_gets_unique_uuid7(self, mixin_session):
+        """Each inserted row gets a distinct UUIDv7."""
+        a = UUIDv7Model(name="a")
+        b = UUIDv7Model(name="b")
+        mixin_session.add_all([a, b])
+        await mixin_session.flush()
+
+        assert a.id != b.id
+
+    @pytest.mark.anyio
+    async def test_uuid7_version(self, mixin_session):
+        """Generated UUIDs have version 7."""
+        obj = UUIDv7Model(name="test")
+        mixin_session.add(obj)
+        await mixin_session.flush()
+
+        assert obj.id.version == 7
+
+    @pytest.mark.anyio
+    async def test_uuid7_server_default_set(self):
+        """Column has uuidv7() as server default."""
+        col = UUIDv7Model.__table__.c["id"]
+        assert col.server_default is not None
+        assert "uuidv7" in str(col.server_default.arg)
+
+
 class TestFullMixinModel:
     @pytest.mark.anyio
     async def test_combined_mixins_work_together(self, mixin_session):

tests/test_schemas.py

@@ -6,9 +6,12 @@ from pydantic import ValidationError
 from fastapi_toolsets.schemas import (
     ApiError,
     CursorPagination,
+    CursorPaginatedResponse,
     ErrorResponse,
     OffsetPagination,
+    OffsetPaginatedResponse,
     PaginatedResponse,
+    PaginationType,
     Response,
     ResponseStatus,
 )
@@ -301,7 +304,7 @@ class TestPaginatedResponse:
             page=1,
             has_more=False,
         )
-        response = PaginatedResponse[dict](
+        response = PaginatedResponse(
             data=[],
             pagination=pagination,
         )
@@ -310,13 +313,32 @@ 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."""
-
-        class UserOut:
-            id: int
-            name: str
-
         pagination = OffsetPagination(
             total_count=1,
             items_per_page=10,
@@ -371,6 +393,191 @@ class TestPaginatedResponse:
         assert isinstance(response.pagination, CursorPagination)
 
 
+class TestPaginationType:
+    """Tests for PaginationType enum."""
+
+    def test_offset_value(self):
+        """OFFSET has string value 'offset'."""
+        assert PaginationType.OFFSET == "offset"
+        assert PaginationType.OFFSET.value == "offset"
+
+    def test_cursor_value(self):
+        """CURSOR has string value 'cursor'."""
+        assert PaginationType.CURSOR == "cursor"
+        assert PaginationType.CURSOR.value == "cursor"
+
+    def test_is_string_enum(self):
+        """PaginationType is a string enum."""
+        assert isinstance(PaginationType.OFFSET, str)
+        assert isinstance(PaginationType.CURSOR, str)
+
+    def test_members(self):
+        """PaginationType has exactly two members."""
+        assert set(PaginationType) == {PaginationType.OFFSET, PaginationType.CURSOR}
+
+
+class TestOffsetPaginatedResponse:
+    """Tests for OffsetPaginatedResponse schema."""
+
+    def test_pagination_type_is_offset(self):
+        """pagination_type is always PaginationType.OFFSET."""
+        response = OffsetPaginatedResponse(
+            data=[],
+            pagination=OffsetPagination(
+                total_count=0, items_per_page=10, page=1, has_more=False
+            ),
+        )
+        assert response.pagination_type is PaginationType.OFFSET
+
+    def test_pagination_type_serializes_to_string(self):
+        """pagination_type serializes to 'offset' in JSON mode."""
+        response = OffsetPaginatedResponse(
+            data=[],
+            pagination=OffsetPagination(
+                total_count=0, items_per_page=10, page=1, has_more=False
+            ),
+        )
+        assert response.model_dump(mode="json")["pagination_type"] == "offset"
+
+    def test_pagination_field_is_typed(self):
+        """pagination field is OffsetPagination, not the union."""
+        response = OffsetPaginatedResponse(
+            data=[{"id": 1}],
+            pagination=OffsetPagination(
+                total_count=10, items_per_page=5, page=2, has_more=True
+            ),
+        )
+        assert isinstance(response.pagination, OffsetPagination)
+        assert response.pagination.total_count == 10
+        assert response.pagination.page == 2
+
+    def test_is_subclass_of_paginated_response(self):
+        """OffsetPaginatedResponse IS a PaginatedResponse."""
+        response = OffsetPaginatedResponse(
+            data=[],
+            pagination=OffsetPagination(
+                total_count=0, items_per_page=10, page=1, has_more=False
+            ),
+        )
+        assert isinstance(response, PaginatedResponse)
+
+    def test_pagination_type_default_cannot_be_overridden_to_cursor(self):
+        """pagination_type rejects values other than OFFSET."""
+        with pytest.raises(ValidationError):
+            OffsetPaginatedResponse(
+                data=[],
+                pagination=OffsetPagination(
+                    total_count=0, items_per_page=10, page=1, has_more=False
+                ),
+                pagination_type=PaginationType.CURSOR,  # type: ignore[arg-type]
+            )
+
+    def test_filter_attributes_defaults_to_none(self):
+        """filter_attributes defaults to None."""
+        response = OffsetPaginatedResponse(
+            data=[],
+            pagination=OffsetPagination(
+                total_count=0, items_per_page=10, page=1, has_more=False
+            ),
+        )
+        assert response.filter_attributes is None
+
+    def test_full_serialization(self):
+        """Full JSON serialization includes all expected fields."""
+        response = OffsetPaginatedResponse(
+            data=[{"id": 1}],
+            pagination=OffsetPagination(
+                total_count=1, items_per_page=10, page=1, has_more=False
+            ),
+            filter_attributes={"status": ["active"]},
+        )
+        data = response.model_dump(mode="json")
+
+        assert data["pagination_type"] == "offset"
+        assert data["status"] == "SUCCESS"
+        assert data["data"] == [{"id": 1}]
+        assert data["pagination"]["total_count"] == 1
+        assert data["filter_attributes"] == {"status": ["active"]}
+
+
+class TestCursorPaginatedResponse:
+    """Tests for CursorPaginatedResponse schema."""
+
+    def test_pagination_type_is_cursor(self):
+        """pagination_type is always PaginationType.CURSOR."""
+        response = CursorPaginatedResponse(
+            data=[],
+            pagination=CursorPagination(
+                next_cursor=None, items_per_page=10, has_more=False
+            ),
+        )
+        assert response.pagination_type is PaginationType.CURSOR
+
+    def test_pagination_type_serializes_to_string(self):
+        """pagination_type serializes to 'cursor' in JSON mode."""
+        response = CursorPaginatedResponse(
+            data=[],
+            pagination=CursorPagination(
+                next_cursor=None, items_per_page=10, has_more=False
+            ),
+        )
+        assert response.model_dump(mode="json")["pagination_type"] == "cursor"
+
+    def test_pagination_field_is_typed(self):
+        """pagination field is CursorPagination, not the union."""
+        response = CursorPaginatedResponse(
+            data=[{"id": 1}],
+            pagination=CursorPagination(
+                next_cursor="abc123",
+                prev_cursor=None,
+                items_per_page=20,
+                has_more=True,
+            ),
+        )
+        assert isinstance(response.pagination, CursorPagination)
+        assert response.pagination.next_cursor == "abc123"
+        assert response.pagination.has_more is True
+
+    def test_is_subclass_of_paginated_response(self):
+        """CursorPaginatedResponse IS a PaginatedResponse."""
+        response = CursorPaginatedResponse(
+            data=[],
+            pagination=CursorPagination(
+                next_cursor=None, items_per_page=10, has_more=False
+            ),
+        )
+        assert isinstance(response, PaginatedResponse)
+
+    def test_pagination_type_default_cannot_be_overridden_to_offset(self):
+        """pagination_type rejects values other than CURSOR."""
+        with pytest.raises(ValidationError):
+            CursorPaginatedResponse(
+                data=[],
+                pagination=CursorPagination(
+                    next_cursor=None, items_per_page=10, has_more=False
+                ),
+                pagination_type=PaginationType.OFFSET,  # type: ignore[arg-type]
+            )
+
+    def test_full_serialization(self):
+        """Full JSON serialization includes all expected fields."""
+        response = CursorPaginatedResponse(
+            data=[{"id": 1}],
+            pagination=CursorPagination(
+                next_cursor="tok_next",
+                prev_cursor="tok_prev",
+                items_per_page=10,
+                has_more=True,
+            ),
+        )
+        data = response.model_dump(mode="json")
+
+        assert data["pagination_type"] == "cursor"
+        assert data["status"] == "SUCCESS"
+        assert data["pagination"]["next_cursor"] == "tok_next"
+        assert data["pagination"]["prev_cursor"] == "tok_prev"
+
+
 class TestFromAttributes:
     """Tests for from_attributes config (ORM mode)."""
 

uv.lock

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