Compare commits

..

13 Commits

Author SHA1 Message Date
8fdf60254b fix: cleanup + simplify 2026-03-18 15:31:04 -04:00
90a244031e docs: add authentication example 2026-03-18 15:13:34 -04:00
f561ccf5a9 feat(security): add oauth helpers 2026-03-18 15:13:34 -04:00
0f40d858fe feat: add security module 2026-03-18 15:13:34 -04:00
d3vyce
2e9c6c0c90 feat: support Annotated[AsyncSession, Depends(...)] in PathDependency and BodyDependency (#146) 2026-03-18 20:10:58 +01:00
2c494fcd17 Version 2.3.0 2026-03-15 12:52:44 -04:00
d3vyce
fd7269a372 refactor: CursorDirection enum, cursor value parsing and __class_getitem__ caching (#144) 2026-03-15 17:48:50 +01:00
d3vyce
c863744012 feat: PaginatedResponse[T] as discriminated union annotation (#142) 2026-03-15 17:41:33 +01:00
d3vyce
aedcbf4e04 feat: add UUIDv7Mixin (#140) 2026-03-14 20:41:46 +01:00
d3vyce
19c013bdec feat: unified paginate() endpoint with typed pagination responses (#134)
* feat: unified paginate() endpoint with typed pagination responses

* docs: unified paginate() endpoint

* fix: add tests
2026-03-14 20:23:20 +01:00
d3vyce
81407c3038 fix: use clock_timestamp() instead of now() for Mixin to ensure unique values (#138) 2026-03-14 17:30:11 +01:00
d3vyce
0fb00d44da fix: prev_cursor does not navigate backward (#136) 2026-03-14 17:18:53 +01:00
d3vyce
19232d3436 feat: add AsyncCrud subclass style and base_class param to CrudFactory (#132) 2026-03-12 22:46:51 +01:00
45 changed files with 4091 additions and 178 deletions

View File

@@ -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`

View File

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

View File

@@ -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`:

View File

@@ -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`

View File

@@ -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,

View File

@@ -13,8 +13,13 @@ The `dependencies` module provides two factory functions that create FastAPI dep
```python
from fastapi_toolsets.dependencies import PathDependency
# Plain callable
UserDep = PathDependency(model=User, field=User.id, session_dep=get_db)
# Annotated
SessionDep = Annotated[AsyncSession, Depends(get_db)]
UserDep = PathDependency(model=User, field=User.id, session_dep=SessionDep)
@router.get("/users/{user_id}")
async def get_user(user: User = UserDep):
return user
@@ -37,8 +42,14 @@ async def get_user(user: User = UserDep):
```python
from fastapi_toolsets.dependencies import BodyDependency
# Plain callable
RoleDep = BodyDependency(model=Role, field=Role.id, session_dep=get_db, body_field="role_id")
# Annotated
SessionDep = Annotated[AsyncSession, Depends(get_db)]
RoleDep = BodyDependency(model=Role, field=Role.id, session_dep=SessionDep, body_field="role_id")
@router.post("/users")
async def create_user(body: UserCreateSchema, role: Role = RoleDep):
user = User(username=body.username, role=role)

View File

@@ -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

View File

@@ -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)

267
docs/module/security.md Normal file
View File

@@ -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)

View File

@@ -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

View File

@@ -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

View File

@@ -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

View File

@@ -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)

View File

@@ -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)

View File

@@ -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)

View File

@@ -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")

View File

@@ -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],
)

View File

@@ -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

View File

@@ -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

View File

@@ -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))

View File

@@ -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,
)

View File

@@ -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"
@@ -66,6 +66,7 @@ manager = "fastapi_toolsets.cli.app:cli"
dev = [
{include-group = "tests"},
{include-group = "docs"},
{include-group = "docs-src"},
"fastapi-toolsets[all]",
"ruff>=0.1.0",
"ty>=0.0.1a0",
@@ -82,6 +83,9 @@ docs = [
"mkdocstrings-python>=2.0.2",
"zensical>=0.0.23",
]
docs-src = [
"bcrypt>=4.0.0",
]
[build-system]
requires = ["uv_build>=0.10,<0.11.0"]

View File

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

View File

@@ -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",
]

View File

@@ -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,25 +1076,14 @@ 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)
# Build search filters
@@ -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
# 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,

View File

@@ -1,10 +1,12 @@
"""Dependency factories for FastAPI routes."""
import inspect
import typing
from collections.abc import Callable
from typing import Any, cast
from fastapi import Depends
from fastapi.params import Depends as DependsClass
from sqlalchemy.ext.asyncio import AsyncSession
from .crud import CrudFactory
@@ -13,6 +15,15 @@ from .types import ModelType, SessionDependency
__all__ = ["BodyDependency", "PathDependency"]
def _unwrap_session_dep(session_dep: SessionDependency) -> Callable[..., Any]:
"""Extract the plain callable from ``Annotated[AsyncSession, Depends(fn)]`` if needed."""
if typing.get_origin(session_dep) is typing.Annotated:
for arg in typing.get_args(session_dep)[1:]:
if isinstance(arg, DependsClass):
return arg.dependency
return session_dep
def PathDependency(
model: type[ModelType],
field: Any,
@@ -44,6 +55,7 @@ def PathDependency(
): ...
```
"""
session_callable = _unwrap_session_dep(session_dep)
crud = CrudFactory(model)
name = (
param_name
@@ -53,7 +65,7 @@ def PathDependency(
python_type = field.type.python_type
async def dependency(
session: AsyncSession = Depends(session_dep), **kwargs: Any
session: AsyncSession = Depends(session_callable), **kwargs: Any
) -> ModelType:
value = kwargs[name]
return await crud.get(session, filters=[field == value])
@@ -70,7 +82,7 @@ def PathDependency(
"session",
inspect.Parameter.KEYWORD_ONLY,
annotation=AsyncSession,
default=Depends(session_dep),
default=Depends(session_callable),
),
]
),
@@ -112,11 +124,12 @@ def BodyDependency(
): ...
```
"""
session_callable = _unwrap_session_dep(session_dep)
crud = CrudFactory(model)
python_type = field.type.python_type
async def dependency(
session: AsyncSession = Depends(session_dep), **kwargs: Any
session: AsyncSession = Depends(session_callable), **kwargs: Any
) -> ModelType:
value = kwargs[body_field]
return await crud.get(session, filters=[field == value])
@@ -133,7 +146,7 @@ def BodyDependency(
"session",
inspect.Parameter.KEYWORD_ONLY,
annotation=AsyncSession,
default=Depends(session_dep),
default=Depends(session_callable),
),
]
),

View File

@@ -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()"),
)

View File

@@ -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

View File

@@ -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",
]

View File

@@ -0,0 +1,53 @@
"""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
def _ensure_async(fn: Callable[..., Any]) -> Callable[..., Any]:
"""Wrap *fn* so it can always be awaited, caching the coroutine check at init time."""
if inspect.iscoroutinefunction(fn):
return fn
async def wrapper(*args: Any, **kwargs: Any) -> Any:
return fn(*args, **kwargs)
return wrapper
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)

View File

@@ -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

View File

@@ -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"]

View File

@@ -0,0 +1,120 @@
"""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, _ensure_async
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 = _ensure_async(validator)
self._prefix = prefix
self._kwargs = kwargs
self._scheme = HTTPBearer(auto_error=False)
async def _call(
security_scopes: SecurityScopes, # noqa: ARG001
credentials: Annotated[
HTTPAuthorizationCredentials | None, Depends(self._scheme)
] = None,
) -> Any:
if credentials is None:
raise UnauthorizedError()
return await self._validate(credentials.credentials)
self._call_fn = _call
self.__signature__ = inspect.signature(_call)
async def _validate(self, token: str) -> Any:
"""Check prefix and call the validator."""
if self._prefix is not None and not token.startswith(self._prefix):
raise UnauthorizedError()
return await self._validator(token, **self._kwargs)
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 self._validate(credential)
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

View File

@@ -0,0 +1,139 @@
"""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, _ensure_async
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 = _ensure_async(validator)
self._secret_key = secret_key
self._ttl = ttl
self._kwargs = kwargs
self._scheme = APIKeyCookie(name=name, auto_error=False)
async def _call(
security_scopes: SecurityScopes, # noqa: ARG001
value: Annotated[str | None, Depends(self._scheme)] = None,
) -> Any:
if value is None:
raise UnauthorizedError()
plain = self._verify(value)
return await self._validator(plain, **self._kwargs)
self._call_fn = _call
self.__signature__ = inspect.signature(_call)
def _hmac(self, data: str) -> str:
if self._secret_key is None:
raise RuntimeError("_hmac called without secret_key configured")
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 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")

View File

@@ -0,0 +1,67 @@
"""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, _ensure_async
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 = _ensure_async(validator)
self._kwargs = kwargs
self._scheme = APIKeyHeader(name=name, auto_error=False)
async def _call(
security_scopes: SecurityScopes, # noqa: ARG001
api_key: Annotated[str | None, Depends(self._scheme)] = None,
) -> Any:
if api_key is None:
raise UnauthorizedError()
return await self._validator(api_key, **self._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 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},
)

View File

@@ -0,0 +1,119 @@
"""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
async def _call(
request: Request,
security_scopes: SecurityScopes, # noqa: ARG001
**kwargs: Any, # noqa: ARG001 — absorbs scheme values injected by FastAPI
) -> Any:
for source in self._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)

View File

@@ -24,4 +24,4 @@ SearchFieldType = InstrumentedAttribute[Any] | tuple[InstrumentedAttribute[Any],
FacetFieldType = SearchFieldType
# Dependency type aliases
SessionDependency = Callable[[], AsyncGenerator[AsyncSession, None]]
SessionDependency = Callable[[], AsyncGenerator[AsyncSession, None]] | Any

View File

@@ -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]

View File

@@ -3,13 +3,17 @@
import inspect
import uuid
from collections.abc import AsyncGenerator
from typing import Any, cast
from typing import Annotated, Any, cast
import pytest
from fastapi.params import Depends
from sqlalchemy.ext.asyncio import AsyncSession
from fastapi_toolsets.dependencies import BodyDependency, PathDependency
from fastapi_toolsets.dependencies import (
BodyDependency,
PathDependency,
_unwrap_session_dep,
)
from .conftest import Role, RoleCreate, RoleCrud, User
@@ -19,6 +23,24 @@ async def mock_get_db() -> AsyncGenerator[AsyncSession, None]:
yield None
MockSessionDep = Annotated[AsyncSession, Depends(mock_get_db)]
class TestUnwrapSessionDep:
def test_plain_callable_returned_as_is(self):
"""Plain callable is returned unchanged."""
assert _unwrap_session_dep(mock_get_db) is mock_get_db
def test_annotated_with_depends_unwrapped(self):
"""Annotated form with Depends is unwrapped to the plain callable."""
assert _unwrap_session_dep(MockSessionDep) is mock_get_db
def test_annotated_without_depends_returned_as_is(self):
"""Annotated form with no Depends falls back to returning session_dep as-is."""
annotated_no_dep = Annotated[AsyncSession, "not_a_depends"]
assert _unwrap_session_dep(annotated_no_dep) is annotated_no_dep
class TestPathDependency:
"""Tests for PathDependency factory."""
@@ -95,6 +117,39 @@ class TestPathDependency:
assert result.id == role.id
assert result.name == "test_role"
def test_annotated_session_dep_returns_depends_instance(self):
"""PathDependency accepts Annotated[AsyncSession, Depends(...)] form."""
dep = PathDependency(Role, Role.id, session_dep=MockSessionDep)
assert isinstance(dep, Depends)
def test_annotated_session_dep_signature(self):
"""PathDependency with Annotated session_dep produces a valid signature."""
dep = cast(Any, PathDependency(Role, Role.id, session_dep=MockSessionDep))
sig = inspect.signature(dep.dependency)
assert "role_id" in sig.parameters
assert "session" in sig.parameters
assert isinstance(sig.parameters["session"].default, Depends)
def test_annotated_session_dep_unwraps_callable(self):
"""PathDependency with Annotated form uses the underlying callable, not the Annotated type."""
dep = cast(Any, PathDependency(Role, Role.id, session_dep=MockSessionDep))
sig = inspect.signature(dep.dependency)
inner_dep = sig.parameters["session"].default
assert inner_dep.dependency is mock_get_db
@pytest.mark.anyio
async def test_annotated_session_dep_fetches_object(self, db_session):
"""PathDependency with Annotated session_dep correctly fetches object from database."""
role = await RoleCrud.create(db_session, RoleCreate(name="annotated_role"))
dep = cast(Any, PathDependency(Role, Role.id, session_dep=MockSessionDep))
result = await dep.dependency(session=db_session, role_id=role.id)
assert result.id == role.id
assert result.name == "annotated_role"
class TestBodyDependency:
"""Tests for BodyDependency factory."""
@@ -184,3 +239,39 @@ class TestBodyDependency:
assert result.id == role.id
assert result.name == "body_test_role"
def test_annotated_session_dep_returns_depends_instance(self):
"""BodyDependency accepts Annotated[AsyncSession, Depends(...)] form."""
dep = BodyDependency(
Role, Role.id, session_dep=MockSessionDep, body_field="role_id"
)
assert isinstance(dep, Depends)
def test_annotated_session_dep_unwraps_callable(self):
"""BodyDependency with Annotated form uses the underlying callable, not the Annotated type."""
dep = cast(
Any,
BodyDependency(
Role, Role.id, session_dep=MockSessionDep, body_field="role_id"
),
)
sig = inspect.signature(dep.dependency)
inner_dep = sig.parameters["session"].default
assert inner_dep.dependency is mock_get_db
@pytest.mark.anyio
async def test_annotated_session_dep_fetches_object(self, db_session):
"""BodyDependency with Annotated session_dep correctly fetches object from database."""
role = await RoleCrud.create(db_session, RoleCreate(name="body_annotated_role"))
dep = cast(
Any,
BodyDependency(
Role, Role.id, session_dep=MockSessionDep, body_field="role_id"
),
)
result = await dep.dependency(session=db_session, role_id=role.id)
assert result.id == role.id
assert result.name == "body_annotated_role"

View File

@@ -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

View File

@@ -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):

View File

@@ -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)."""

1180
tests/test_security.py Normal file

File diff suppressed because it is too large Load Diff

78
uv.lock generated
View File

@@ -81,6 +81,76 @@ wheels = [
{ url = "https://files.pythonhosted.org/packages/3c/d7/8fb3044eaef08a310acfe23dae9a8e2e07d305edc29a53497e52bc76eca7/asyncpg-0.31.0-cp314-cp314t-win_amd64.whl", hash = "sha256:bd4107bb7cdd0e9e65fae66a62afd3a249663b844fa34d479f6d5b3bef9c04c3", size = 706062, upload-time = "2025-11-24T23:26:44.086Z" },
]
[[package]]
name = "bcrypt"
version = "5.0.0"
source = { registry = "https://pypi.org/simple" }
sdist = { url = "https://files.pythonhosted.org/packages/d4/36/3329e2518d70ad8e2e5817d5a4cac6bba05a47767ec416c7d020a965f408/bcrypt-5.0.0.tar.gz", hash = "sha256:f748f7c2d6fd375cc93d3fba7ef4a9e3a092421b8dbf34d8d4dc06be9492dfdd", size = 25386, upload-time = "2025-09-25T19:50:47.829Z" }
wheels = [
{ url = "https://files.pythonhosted.org/packages/13/85/3e65e01985fddf25b64ca67275bb5bdb4040bd1a53b66d355c6c37c8a680/bcrypt-5.0.0-cp313-cp313t-macosx_10_12_universal2.whl", hash = "sha256:f3c08197f3039bec79cee59a606d62b96b16669cff3949f21e74796b6e3cd2be", size = 481806, upload-time = "2025-09-25T19:49:05.102Z" },
{ url = "https://files.pythonhosted.org/packages/44/dc/01eb79f12b177017a726cbf78330eb0eb442fae0e7b3dfd84ea2849552f3/bcrypt-5.0.0-cp313-cp313t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:200af71bc25f22006f4069060c88ed36f8aa4ff7f53e67ff04d2ab3f1e79a5b2", size = 268626, upload-time = "2025-09-25T19:49:06.723Z" },
{ url = "https://files.pythonhosted.org/packages/8c/cf/e82388ad5959c40d6afd94fb4743cc077129d45b952d46bdc3180310e2df/bcrypt-5.0.0-cp313-cp313t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:baade0a5657654c2984468efb7d6c110db87ea63ef5a4b54732e7e337253e44f", size = 271853, upload-time = "2025-09-25T19:49:08.028Z" },
{ url = "https://files.pythonhosted.org/packages/ec/86/7134b9dae7cf0efa85671651341f6afa695857fae172615e960fb6a466fa/bcrypt-5.0.0-cp313-cp313t-manylinux_2_28_aarch64.whl", hash = "sha256:c58b56cdfb03202b3bcc9fd8daee8e8e9b6d7e3163aa97c631dfcfcc24d36c86", size = 269793, upload-time = "2025-09-25T19:49:09.727Z" },
{ url = "https://files.pythonhosted.org/packages/cc/82/6296688ac1b9e503d034e7d0614d56e80c5d1a08402ff856a4549cb59207/bcrypt-5.0.0-cp313-cp313t-manylinux_2_28_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:4bfd2a34de661f34d0bda43c3e4e79df586e4716ef401fe31ea39d69d581ef23", size = 289930, upload-time = "2025-09-25T19:49:11.204Z" },
{ url = "https://files.pythonhosted.org/packages/d1/18/884a44aa47f2a3b88dd09bc05a1e40b57878ecd111d17e5bba6f09f8bb77/bcrypt-5.0.0-cp313-cp313t-manylinux_2_28_x86_64.whl", hash = "sha256:ed2e1365e31fc73f1825fa830f1c8f8917ca1b3ca6185773b349c20fd606cec2", size = 272194, upload-time = "2025-09-25T19:49:12.524Z" },
{ url = "https://files.pythonhosted.org/packages/0e/8f/371a3ab33c6982070b674f1788e05b656cfbf5685894acbfef0c65483a59/bcrypt-5.0.0-cp313-cp313t-manylinux_2_34_aarch64.whl", hash = "sha256:83e787d7a84dbbfba6f250dd7a5efd689e935f03dd83b0f919d39349e1f23f83", size = 269381, upload-time = "2025-09-25T19:49:14.308Z" },
{ url = "https://files.pythonhosted.org/packages/b1/34/7e4e6abb7a8778db6422e88b1f06eb07c47682313997ee8a8f9352e5a6f1/bcrypt-5.0.0-cp313-cp313t-manylinux_2_34_x86_64.whl", hash = "sha256:137c5156524328a24b9fac1cb5db0ba618bc97d11970b39184c1d87dc4bf1746", size = 271750, upload-time = "2025-09-25T19:49:15.584Z" },
{ url = "https://files.pythonhosted.org/packages/c0/1b/54f416be2499bd72123c70d98d36c6cd61a4e33d9b89562c22481c81bb30/bcrypt-5.0.0-cp313-cp313t-musllinux_1_1_aarch64.whl", hash = "sha256:38cac74101777a6a7d3b3e3cfefa57089b5ada650dce2baf0cbdd9d65db22a9e", size = 303757, upload-time = "2025-09-25T19:49:17.244Z" },
{ url = "https://files.pythonhosted.org/packages/13/62/062c24c7bcf9d2826a1a843d0d605c65a755bc98002923d01fd61270705a/bcrypt-5.0.0-cp313-cp313t-musllinux_1_1_x86_64.whl", hash = "sha256:d8d65b564ec849643d9f7ea05c6d9f0cd7ca23bdd4ac0c2dbef1104ab504543d", size = 306740, upload-time = "2025-09-25T19:49:18.693Z" },
{ url = "https://files.pythonhosted.org/packages/d5/c8/1fdbfc8c0f20875b6b4020f3c7dc447b8de60aa0be5faaf009d24242aec9/bcrypt-5.0.0-cp313-cp313t-musllinux_1_2_aarch64.whl", hash = "sha256:741449132f64b3524e95cd30e5cd3343006ce146088f074f31ab26b94e6c75ba", size = 334197, upload-time = "2025-09-25T19:49:20.523Z" },
{ url = "https://files.pythonhosted.org/packages/a6/c1/8b84545382d75bef226fbc6588af0f7b7d095f7cd6a670b42a86243183cd/bcrypt-5.0.0-cp313-cp313t-musllinux_1_2_x86_64.whl", hash = "sha256:212139484ab3207b1f0c00633d3be92fef3c5f0af17cad155679d03ff2ee1e41", size = 352974, upload-time = "2025-09-25T19:49:22.254Z" },
{ url = "https://files.pythonhosted.org/packages/10/a6/ffb49d4254ed085e62e3e5dd05982b4393e32fe1e49bb1130186617c29cd/bcrypt-5.0.0-cp313-cp313t-win32.whl", hash = "sha256:9d52ed507c2488eddd6a95bccee4e808d3234fa78dd370e24bac65a21212b861", size = 148498, upload-time = "2025-09-25T19:49:24.134Z" },
{ url = "https://files.pythonhosted.org/packages/48/a9/259559edc85258b6d5fc5471a62a3299a6aa37a6611a169756bf4689323c/bcrypt-5.0.0-cp313-cp313t-win_amd64.whl", hash = "sha256:f6984a24db30548fd39a44360532898c33528b74aedf81c26cf29c51ee47057e", size = 145853, upload-time = "2025-09-25T19:49:25.702Z" },
{ url = "https://files.pythonhosted.org/packages/2d/df/9714173403c7e8b245acf8e4be8876aac64a209d1b392af457c79e60492e/bcrypt-5.0.0-cp313-cp313t-win_arm64.whl", hash = "sha256:9fffdb387abe6aa775af36ef16f55e318dcda4194ddbf82007a6f21da29de8f5", size = 139626, upload-time = "2025-09-25T19:49:26.928Z" },
{ url = "https://files.pythonhosted.org/packages/f8/14/c18006f91816606a4abe294ccc5d1e6f0e42304df5a33710e9e8e95416e1/bcrypt-5.0.0-cp314-cp314t-macosx_10_12_universal2.whl", hash = "sha256:4870a52610537037adb382444fefd3706d96d663ac44cbb2f37e3919dca3d7ef", size = 481862, upload-time = "2025-09-25T19:49:28.365Z" },
{ url = "https://files.pythonhosted.org/packages/67/49/dd074d831f00e589537e07a0725cf0e220d1f0d5d8e85ad5bbff251c45aa/bcrypt-5.0.0-cp314-cp314t-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:48f753100931605686f74e27a7b49238122aa761a9aefe9373265b8b7aa43ea4", size = 268544, upload-time = "2025-09-25T19:49:30.39Z" },
{ url = "https://files.pythonhosted.org/packages/f5/91/50ccba088b8c474545b034a1424d05195d9fcbaaf802ab8bfe2be5a4e0d7/bcrypt-5.0.0-cp314-cp314t-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:f70aadb7a809305226daedf75d90379c397b094755a710d7014b8b117df1ebbf", size = 271787, upload-time = "2025-09-25T19:49:32.144Z" },
{ url = "https://files.pythonhosted.org/packages/aa/e7/d7dba133e02abcda3b52087a7eea8c0d4f64d3e593b4fffc10c31b7061f3/bcrypt-5.0.0-cp314-cp314t-manylinux_2_28_aarch64.whl", hash = "sha256:744d3c6b164caa658adcb72cb8cc9ad9b4b75c7db507ab4bc2480474a51989da", size = 269753, upload-time = "2025-09-25T19:49:33.885Z" },
{ url = "https://files.pythonhosted.org/packages/33/fc/5b145673c4b8d01018307b5c2c1fc87a6f5a436f0ad56607aee389de8ee3/bcrypt-5.0.0-cp314-cp314t-manylinux_2_28_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:a28bc05039bdf3289d757f49d616ab3efe8cf40d8e8001ccdd621cd4f98f4fc9", size = 289587, upload-time = "2025-09-25T19:49:35.144Z" },
{ url = "https://files.pythonhosted.org/packages/27/d7/1ff22703ec6d4f90e62f1a5654b8867ef96bafb8e8102c2288333e1a6ca6/bcrypt-5.0.0-cp314-cp314t-manylinux_2_28_x86_64.whl", hash = "sha256:7f277a4b3390ab4bebe597800a90da0edae882c6196d3038a73adf446c4f969f", size = 272178, upload-time = "2025-09-25T19:49:36.793Z" },
{ url = "https://files.pythonhosted.org/packages/c8/88/815b6d558a1e4d40ece04a2f84865b0fef233513bd85fd0e40c294272d62/bcrypt-5.0.0-cp314-cp314t-manylinux_2_34_aarch64.whl", hash = "sha256:79cfa161eda8d2ddf29acad370356b47f02387153b11d46042e93a0a95127493", size = 269295, upload-time = "2025-09-25T19:49:38.164Z" },
{ url = "https://files.pythonhosted.org/packages/51/8c/e0db387c79ab4931fc89827d37608c31cc57b6edc08ccd2386139028dc0d/bcrypt-5.0.0-cp314-cp314t-manylinux_2_34_x86_64.whl", hash = "sha256:a5393eae5722bcef046a990b84dff02b954904c36a194f6cfc817d7dca6c6f0b", size = 271700, upload-time = "2025-09-25T19:49:39.917Z" },
{ url = "https://files.pythonhosted.org/packages/06/83/1570edddd150f572dbe9fc00f6203a89fc7d4226821f67328a85c330f239/bcrypt-5.0.0-cp314-cp314t-musllinux_1_2_aarch64.whl", hash = "sha256:7f4c94dec1b5ab5d522750cb059bb9409ea8872d4494fd152b53cca99f1ddd8c", size = 334034, upload-time = "2025-09-25T19:49:41.227Z" },
{ url = "https://files.pythonhosted.org/packages/c9/f2/ea64e51a65e56ae7a8a4ec236c2bfbdd4b23008abd50ac33fbb2d1d15424/bcrypt-5.0.0-cp314-cp314t-musllinux_1_2_x86_64.whl", hash = "sha256:0cae4cb350934dfd74c020525eeae0a5f79257e8a201c0c176f4b84fdbf2a4b4", size = 352766, upload-time = "2025-09-25T19:49:43.08Z" },
{ url = "https://files.pythonhosted.org/packages/d7/d4/1a388d21ee66876f27d1a1f41287897d0c0f1712ef97d395d708ba93004c/bcrypt-5.0.0-cp314-cp314t-win32.whl", hash = "sha256:b17366316c654e1ad0306a6858e189fc835eca39f7eb2cafd6aaca8ce0c40a2e", size = 152449, upload-time = "2025-09-25T19:49:44.971Z" },
{ url = "https://files.pythonhosted.org/packages/3f/61/3291c2243ae0229e5bca5d19f4032cecad5dfb05a2557169d3a69dc0ba91/bcrypt-5.0.0-cp314-cp314t-win_amd64.whl", hash = "sha256:92864f54fb48b4c718fc92a32825d0e42265a627f956bc0361fe869f1adc3e7d", size = 149310, upload-time = "2025-09-25T19:49:46.162Z" },
{ url = "https://files.pythonhosted.org/packages/3e/89/4b01c52ae0c1a681d4021e5dd3e45b111a8fb47254a274fa9a378d8d834b/bcrypt-5.0.0-cp314-cp314t-win_arm64.whl", hash = "sha256:dd19cf5184a90c873009244586396a6a884d591a5323f0e8a5922560718d4993", size = 143761, upload-time = "2025-09-25T19:49:47.345Z" },
{ url = "https://files.pythonhosted.org/packages/84/29/6237f151fbfe295fe3e074ecc6d44228faa1e842a81f6d34a02937ee1736/bcrypt-5.0.0-cp38-abi3-macosx_10_12_universal2.whl", hash = "sha256:fc746432b951e92b58317af8e0ca746efe93e66555f1b40888865ef5bf56446b", size = 494553, upload-time = "2025-09-25T19:49:49.006Z" },
{ url = "https://files.pythonhosted.org/packages/45/b6/4c1205dde5e464ea3bd88e8742e19f899c16fa8916fb8510a851fae985b5/bcrypt-5.0.0-cp38-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:c2388ca94ffee269b6038d48747f4ce8df0ffbea43f31abfa18ac72f0218effb", size = 275009, upload-time = "2025-09-25T19:49:50.581Z" },
{ url = "https://files.pythonhosted.org/packages/3b/71/427945e6ead72ccffe77894b2655b695ccf14ae1866cd977e185d606dd2f/bcrypt-5.0.0-cp38-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:560ddb6ec730386e7b3b26b8b4c88197aaed924430e7b74666a586ac997249ef", size = 278029, upload-time = "2025-09-25T19:49:52.533Z" },
{ url = "https://files.pythonhosted.org/packages/17/72/c344825e3b83c5389a369c8a8e58ffe1480b8a699f46c127c34580c4666b/bcrypt-5.0.0-cp38-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:d79e5c65dcc9af213594d6f7f1fa2c98ad3fc10431e7aa53c176b441943efbdd", size = 275907, upload-time = "2025-09-25T19:49:54.709Z" },
{ url = "https://files.pythonhosted.org/packages/0b/7e/d4e47d2df1641a36d1212e5c0514f5291e1a956a7749f1e595c07a972038/bcrypt-5.0.0-cp38-abi3-manylinux_2_28_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:2b732e7d388fa22d48920baa267ba5d97cca38070b69c0e2d37087b381c681fd", size = 296500, upload-time = "2025-09-25T19:49:56.013Z" },
{ url = "https://files.pythonhosted.org/packages/0f/c3/0ae57a68be2039287ec28bc463b82e4b8dc23f9d12c0be331f4782e19108/bcrypt-5.0.0-cp38-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:0c8e093ea2532601a6f686edbc2c6b2ec24131ff5c52f7610dd64fa4553b5464", size = 278412, upload-time = "2025-09-25T19:49:57.356Z" },
{ url = "https://files.pythonhosted.org/packages/45/2b/77424511adb11e6a99e3a00dcc7745034bee89036ad7d7e255a7e47be7d8/bcrypt-5.0.0-cp38-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:5b1589f4839a0899c146e8892efe320c0fa096568abd9b95593efac50a87cb75", size = 275486, upload-time = "2025-09-25T19:49:59.116Z" },
{ url = "https://files.pythonhosted.org/packages/43/0a/405c753f6158e0f3f14b00b462d8bca31296f7ecfc8fc8bc7919c0c7d73a/bcrypt-5.0.0-cp38-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:89042e61b5e808b67daf24a434d89bab164d4de1746b37a8d173b6b14f3db9ff", size = 277940, upload-time = "2025-09-25T19:50:00.869Z" },
{ url = "https://files.pythonhosted.org/packages/62/83/b3efc285d4aadc1fa83db385ec64dcfa1707e890eb42f03b127d66ac1b7b/bcrypt-5.0.0-cp38-abi3-musllinux_1_1_aarch64.whl", hash = "sha256:e3cf5b2560c7b5a142286f69bde914494b6d8f901aaa71e453078388a50881c4", size = 310776, upload-time = "2025-09-25T19:50:02.393Z" },
{ url = "https://files.pythonhosted.org/packages/95/7d/47ee337dacecde6d234890fe929936cb03ebc4c3a7460854bbd9c97780b8/bcrypt-5.0.0-cp38-abi3-musllinux_1_1_x86_64.whl", hash = "sha256:f632fd56fc4e61564f78b46a2269153122db34988e78b6be8b32d28507b7eaeb", size = 312922, upload-time = "2025-09-25T19:50:04.232Z" },
{ url = "https://files.pythonhosted.org/packages/d6/3a/43d494dfb728f55f4e1cf8fd435d50c16a2d75493225b54c8d06122523c6/bcrypt-5.0.0-cp38-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:801cad5ccb6b87d1b430f183269b94c24f248dddbbc5c1f78b6ed231743e001c", size = 341367, upload-time = "2025-09-25T19:50:05.559Z" },
{ url = "https://files.pythonhosted.org/packages/55/ab/a0727a4547e383e2e22a630e0f908113db37904f58719dc48d4622139b5c/bcrypt-5.0.0-cp38-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:3cf67a804fc66fc217e6914a5635000259fbbbb12e78a99488e4d5ba445a71eb", size = 359187, upload-time = "2025-09-25T19:50:06.916Z" },
{ url = "https://files.pythonhosted.org/packages/1b/bb/461f352fdca663524b4643d8b09e8435b4990f17fbf4fea6bc2a90aa0cc7/bcrypt-5.0.0-cp38-abi3-win32.whl", hash = "sha256:3abeb543874b2c0524ff40c57a4e14e5d3a66ff33fb423529c88f180fd756538", size = 153752, upload-time = "2025-09-25T19:50:08.515Z" },
{ url = "https://files.pythonhosted.org/packages/41/aa/4190e60921927b7056820291f56fc57d00d04757c8b316b2d3c0d1d6da2c/bcrypt-5.0.0-cp38-abi3-win_amd64.whl", hash = "sha256:35a77ec55b541e5e583eb3436ffbbf53b0ffa1fa16ca6782279daf95d146dcd9", size = 150881, upload-time = "2025-09-25T19:50:09.742Z" },
{ url = "https://files.pythonhosted.org/packages/54/12/cd77221719d0b39ac0b55dbd39358db1cd1246e0282e104366ebbfb8266a/bcrypt-5.0.0-cp38-abi3-win_arm64.whl", hash = "sha256:cde08734f12c6a4e28dc6755cd11d3bdfea608d93d958fffbe95a7026ebe4980", size = 144931, upload-time = "2025-09-25T19:50:11.016Z" },
{ url = "https://files.pythonhosted.org/packages/5d/ba/2af136406e1c3839aea9ecadc2f6be2bcd1eff255bd451dd39bcf302c47a/bcrypt-5.0.0-cp39-abi3-macosx_10_12_universal2.whl", hash = "sha256:0c418ca99fd47e9c59a301744d63328f17798b5947b0f791e9af3c1c499c2d0a", size = 495313, upload-time = "2025-09-25T19:50:12.309Z" },
{ url = "https://files.pythonhosted.org/packages/ac/ee/2f4985dbad090ace5ad1f7dd8ff94477fe089b5fab2040bd784a3d5f187b/bcrypt-5.0.0-cp39-abi3-manylinux2014_aarch64.manylinux_2_17_aarch64.whl", hash = "sha256:ddb4e1500f6efdd402218ffe34d040a1196c072e07929b9820f363a1fd1f4191", size = 275290, upload-time = "2025-09-25T19:50:13.673Z" },
{ url = "https://files.pythonhosted.org/packages/e4/6e/b77ade812672d15cf50842e167eead80ac3514f3beacac8902915417f8b7/bcrypt-5.0.0-cp39-abi3-manylinux2014_x86_64.manylinux_2_17_x86_64.whl", hash = "sha256:7aeef54b60ceddb6f30ee3db090351ecf0d40ec6e2abf41430997407a46d2254", size = 278253, upload-time = "2025-09-25T19:50:15.089Z" },
{ url = "https://files.pythonhosted.org/packages/36/c4/ed00ed32f1040f7990dac7115f82273e3c03da1e1a1587a778d8cea496d8/bcrypt-5.0.0-cp39-abi3-manylinux_2_28_aarch64.whl", hash = "sha256:f0ce778135f60799d89c9693b9b398819d15f1921ba15fe719acb3178215a7db", size = 276084, upload-time = "2025-09-25T19:50:16.699Z" },
{ url = "https://files.pythonhosted.org/packages/e7/c4/fa6e16145e145e87f1fa351bbd54b429354fd72145cd3d4e0c5157cf4c70/bcrypt-5.0.0-cp39-abi3-manylinux_2_28_armv7l.manylinux_2_31_armv7l.whl", hash = "sha256:a71f70ee269671460b37a449f5ff26982a6f2ba493b3eabdd687b4bf35f875ac", size = 297185, upload-time = "2025-09-25T19:50:18.525Z" },
{ url = "https://files.pythonhosted.org/packages/24/b4/11f8a31d8b67cca3371e046db49baa7c0594d71eb40ac8121e2fc0888db0/bcrypt-5.0.0-cp39-abi3-manylinux_2_28_x86_64.whl", hash = "sha256:f8429e1c410b4073944f03bd778a9e066e7fad723564a52ff91841d278dfc822", size = 278656, upload-time = "2025-09-25T19:50:19.809Z" },
{ url = "https://files.pythonhosted.org/packages/ac/31/79f11865f8078e192847d2cb526e3fa27c200933c982c5b2869720fa5fce/bcrypt-5.0.0-cp39-abi3-manylinux_2_34_aarch64.whl", hash = "sha256:edfcdcedd0d0f05850c52ba3127b1fce70b9f89e0fe5ff16517df7e81fa3cbb8", size = 275662, upload-time = "2025-09-25T19:50:21.567Z" },
{ url = "https://files.pythonhosted.org/packages/d4/8d/5e43d9584b3b3591a6f9b68f755a4da879a59712981ef5ad2a0ac1379f7a/bcrypt-5.0.0-cp39-abi3-manylinux_2_34_x86_64.whl", hash = "sha256:611f0a17aa4a25a69362dcc299fda5c8a3d4f160e2abb3831041feb77393a14a", size = 278240, upload-time = "2025-09-25T19:50:23.305Z" },
{ url = "https://files.pythonhosted.org/packages/89/48/44590e3fc158620f680a978aafe8f87a4c4320da81ed11552f0323aa9a57/bcrypt-5.0.0-cp39-abi3-musllinux_1_1_aarch64.whl", hash = "sha256:db99dca3b1fdc3db87d7c57eac0c82281242d1eabf19dcb8a6b10eb29a2e72d1", size = 311152, upload-time = "2025-09-25T19:50:24.597Z" },
{ url = "https://files.pythonhosted.org/packages/5f/85/e4fbfc46f14f47b0d20493669a625da5827d07e8a88ee460af6cd9768b44/bcrypt-5.0.0-cp39-abi3-musllinux_1_1_x86_64.whl", hash = "sha256:5feebf85a9cefda32966d8171f5db7e3ba964b77fdfe31919622256f80f9cf42", size = 313284, upload-time = "2025-09-25T19:50:26.268Z" },
{ url = "https://files.pythonhosted.org/packages/25/ae/479f81d3f4594456a01ea2f05b132a519eff9ab5768a70430fa1132384b1/bcrypt-5.0.0-cp39-abi3-musllinux_1_2_aarch64.whl", hash = "sha256:3ca8a166b1140436e058298a34d88032ab62f15aae1c598580333dc21d27ef10", size = 341643, upload-time = "2025-09-25T19:50:28.02Z" },
{ url = "https://files.pythonhosted.org/packages/df/d2/36a086dee1473b14276cd6ea7f61aef3b2648710b5d7f1c9e032c29b859f/bcrypt-5.0.0-cp39-abi3-musllinux_1_2_x86_64.whl", hash = "sha256:61afc381250c3182d9078551e3ac3a41da14154fbff647ddf52a769f588c4172", size = 359698, upload-time = "2025-09-25T19:50:31.347Z" },
{ url = "https://files.pythonhosted.org/packages/c0/f6/688d2cd64bfd0b14d805ddb8a565e11ca1fb0fd6817175d58b10052b6d88/bcrypt-5.0.0-cp39-abi3-win32.whl", hash = "sha256:64d7ce196203e468c457c37ec22390f1a61c85c6f0b8160fd752940ccfb3a683", size = 153725, upload-time = "2025-09-25T19:50:34.384Z" },
{ url = "https://files.pythonhosted.org/packages/9f/b9/9d9a641194a730bda138b3dfe53f584d61c58cd5230e37566e83ec2ffa0d/bcrypt-5.0.0-cp39-abi3-win_amd64.whl", hash = "sha256:64ee8434b0da054d830fa8e89e1c8bf30061d539044a39524ff7dec90481e5c2", size = 150912, upload-time = "2025-09-25T19:50:35.69Z" },
{ url = "https://files.pythonhosted.org/packages/27/44/d2ef5e87509158ad2187f4dd0852df80695bb1ee0cfe0a684727b01a69e0/bcrypt-5.0.0-cp39-abi3-win_arm64.whl", hash = "sha256:f2347d3534e76bf50bca5500989d6c1d05ed64b440408057a37673282c654927", size = 144953, upload-time = "2025-09-25T19:50:37.32Z" },
{ url = "https://files.pythonhosted.org/packages/8a/75/4aa9f5a4d40d762892066ba1046000b329c7cd58e888a6db878019b282dc/bcrypt-5.0.0-pp311-pypy311_pp73-manylinux_2_28_aarch64.whl", hash = "sha256:7edda91d5ab52b15636d9c30da87d2cc84f426c72b9dba7a9b4fe142ba11f534", size = 271180, upload-time = "2025-09-25T19:50:38.575Z" },
{ url = "https://files.pythonhosted.org/packages/54/79/875f9558179573d40a9cc743038ac2bf67dfb79cecb1e8b5d70e88c94c3d/bcrypt-5.0.0-pp311-pypy311_pp73-manylinux_2_28_x86_64.whl", hash = "sha256:046ad6db88edb3c5ece4369af997938fb1c19d6a699b9c1b27b0db432faae4c4", size = 273791, upload-time = "2025-09-25T19:50:39.913Z" },
{ url = "https://files.pythonhosted.org/packages/bc/fe/975adb8c216174bf70fc17535f75e85ac06ed5252ea077be10d9cff5ce24/bcrypt-5.0.0-pp311-pypy311_pp73-manylinux_2_34_aarch64.whl", hash = "sha256:dcd58e2b3a908b5ecc9b9df2f0085592506ac2d5110786018ee5e160f28e0911", size = 270746, upload-time = "2025-09-25T19:50:43.306Z" },
{ url = "https://files.pythonhosted.org/packages/e4/f8/972c96f5a2b6c4b3deca57009d93e946bbdbe2241dca9806d502f29dd3ee/bcrypt-5.0.0-pp311-pypy311_pp73-manylinux_2_34_x86_64.whl", hash = "sha256:6b8f520b61e8781efee73cba14e3e8c9556ccfb375623f4f97429544734545b4", size = 273375, upload-time = "2025-09-25T19:50:45.43Z" },
]
[[package]]
name = "certifi"
version = "2026.1.4"
@@ -251,7 +321,7 @@ wheels = [
[[package]]
name = "fastapi-toolsets"
version = "2.2.1"
version = "2.3.0"
source = { editable = "." }
dependencies = [
{ name = "asyncpg" },
@@ -282,6 +352,7 @@ pytest = [
[package.dev-dependencies]
dev = [
{ name = "bcrypt" },
{ name = "coverage" },
{ name = "fastapi-toolsets", extra = ["all"] },
{ name = "httpx" },
@@ -298,6 +369,9 @@ docs = [
{ name = "mkdocstrings-python" },
{ name = "zensical" },
]
docs-src = [
{ name = "bcrypt" },
]
tests = [
{ name = "coverage" },
{ name = "httpx" },
@@ -324,6 +398,7 @@ provides-extras = ["cli", "metrics", "pytest", "all"]
[package.metadata.requires-dev]
dev = [
{ name = "bcrypt", specifier = ">=4.0.0" },
{ name = "coverage", specifier = ">=7.0.0" },
{ name = "fastapi-toolsets", extras = ["all"] },
{ name = "httpx", specifier = ">=0.25.0" },
@@ -340,6 +415,7 @@ docs = [
{ name = "mkdocstrings-python", specifier = ">=2.0.2" },
{ name = "zensical", specifier = ">=0.0.23" },
]
docs-src = [{ name = "bcrypt", specifier = ">=4.0.0" }]
tests = [
{ name = "coverage", specifier = ">=7.0.0" },
{ name = "httpx", specifier = ">=0.25.0" },