mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-04-16 06:36:26 +02:00
Compare commits
5 Commits
06cec296a5
...
cda3c22ae3
| Author | SHA1 | Date | |
|---|---|---|---|
|
cda3c22ae3
|
|||
|
b6970f3286
|
|||
|
fe1e0779de
|
|||
|
|
32059dcb02 | ||
|
|
f027981e80 |
@@ -43,16 +43,16 @@ Declare `searchable_fields`, `facet_fields`, and `order_fields` once on [`CrudFa
|
|||||||
|
|
||||||
## Routes
|
## Routes
|
||||||
|
|
||||||
```python title="routes.py:1:17"
|
```python title="routes.py:1:16"
|
||||||
--8<-- "docs_src/examples/pagination_search/routes.py:1:17"
|
--8<-- "docs_src/examples/pagination_search/routes.py:1:16"
|
||||||
```
|
```
|
||||||
|
|
||||||
### Offset pagination
|
### Offset pagination
|
||||||
|
|
||||||
Best for admin panels or any UI that needs a total item count and numbered pages.
|
Best for admin panels or any UI that needs a total item count and numbered pages.
|
||||||
|
|
||||||
```python title="routes.py:20:40"
|
```python title="routes.py:19:37"
|
||||||
--8<-- "docs_src/examples/pagination_search/routes.py:20:40"
|
--8<-- "docs_src/examples/pagination_search/routes.py:19:37"
|
||||||
```
|
```
|
||||||
|
|
||||||
**Example request**
|
**Example request**
|
||||||
@@ -92,8 +92,8 @@ To skip the `COUNT(*)` query for better performance on large tables, pass `inclu
|
|||||||
|
|
||||||
Best for feeds, infinite scroll, or any high-throughput API where offset performance degrades.
|
Best for feeds, infinite scroll, or any high-throughput API where offset performance degrades.
|
||||||
|
|
||||||
```python title="routes.py:43:63"
|
```python title="routes.py:40:58"
|
||||||
--8<-- "docs_src/examples/pagination_search/routes.py:43:63"
|
--8<-- "docs_src/examples/pagination_search/routes.py:40:58"
|
||||||
```
|
```
|
||||||
|
|
||||||
**Example request**
|
**Example request**
|
||||||
@@ -132,8 +132,8 @@ Pass `next_cursor` as the `cursor` query parameter on the next request to advanc
|
|||||||
|
|
||||||
[`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.
|
[`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"
|
```python title="routes.py:61:79"
|
||||||
--8<-- "docs_src/examples/pagination_search/routes.py:66:90"
|
--8<-- "docs_src/examples/pagination_search/routes.py:61:79"
|
||||||
```
|
```
|
||||||
|
|
||||||
**Offset request** (default)
|
**Offset request** (default)
|
||||||
|
|||||||
@@ -4,6 +4,30 @@ This page covers every breaking change introduced in **v3.0** and the steps requ
|
|||||||
|
|
||||||
---
|
---
|
||||||
|
|
||||||
|
## CRUD
|
||||||
|
|
||||||
|
### Facet keys now always use the full relationship chain
|
||||||
|
|
||||||
|
In `v2`, relationship facet fields used only the terminal column key (e.g. `"name"` for `Role.name`) and only prepended the relationship name when two facet fields shared the same column key. In `v3`, facet keys **always** include the full relationship chain joined by `__`, regardless of collisions.
|
||||||
|
|
||||||
|
=== "Before (`v2`)"
|
||||||
|
|
||||||
|
```
|
||||||
|
User.status -> status
|
||||||
|
(User.role, Role.name) -> name
|
||||||
|
(User.role, Role.permission, Permission.name) -> name
|
||||||
|
```
|
||||||
|
|
||||||
|
=== "Now (`v3`)"
|
||||||
|
|
||||||
|
```
|
||||||
|
User.status -> status
|
||||||
|
(User.role, Role.name) -> role__name
|
||||||
|
(User.role, Role.permission, Permission.name) -> role__permission__name
|
||||||
|
```
|
||||||
|
|
||||||
|
---
|
||||||
|
|
||||||
## Models
|
## Models
|
||||||
|
|
||||||
The lifecycle event system has been rewritten. Callbacks are now registered with a module-level [`listens_for`](../reference/models.md#fastapi_toolsets.models.listens_for) decorator and dispatched by [`EventSession`](../reference/models.md#fastapi_toolsets.models.EventSession), replacing the mixin-based approach from `v2`.
|
The lifecycle event system has been rewritten. Callbacks are now registered with a module-level [`listens_for`](../reference/models.md#fastapi_toolsets.models.listens_for) decorator and dispatched by [`EventSession`](../reference/models.md#fastapi_toolsets.models.EventSession), replacing the mixin-based approach from `v2`.
|
||||||
|
|||||||
@@ -159,18 +159,15 @@ Three pagination methods are available. All return a typed response whose `pagi
|
|||||||
### Offset pagination
|
### Offset pagination
|
||||||
|
|
||||||
```python
|
```python
|
||||||
|
from typing import Annotated
|
||||||
|
from fastapi import Depends
|
||||||
|
|
||||||
@router.get("")
|
@router.get("")
|
||||||
async def get_users(
|
async def get_users(
|
||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
items_per_page: int = 50,
|
params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
|
||||||
page: int = 1,
|
|
||||||
) -> OffsetPaginatedResponse[UserRead]:
|
) -> OffsetPaginatedResponse[UserRead]:
|
||||||
return await UserCrud.offset_paginate(
|
return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)
|
||||||
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 an [`OffsetPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPaginatedResponse):
|
The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_paginate) method returns an [`OffsetPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.OffsetPaginatedResponse):
|
||||||
@@ -194,32 +191,13 @@ The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.Async
|
|||||||
|
|
||||||
!!! info "Added in `v2.4.1`"
|
!!! info "Added in `v2.4.1`"
|
||||||
|
|
||||||
By default `offset_paginate` runs two queries: one for the page items and one `COUNT(*)` for `total_count`. On large tables the `COUNT` can be expensive. Pass `include_total=False` to skip it:
|
By default `offset_paginate` runs two queries: one for the page items and one `COUNT(*)` for `total_count`. On large tables the `COUNT` can be expensive. Pass `include_total=False` to `offset_paginate_params()` to skip it:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
result = await UserCrud.offset_paginate(
|
|
||||||
session=session,
|
|
||||||
page=page,
|
|
||||||
items_per_page=items_per_page,
|
|
||||||
include_total=False,
|
|
||||||
schema=UserRead,
|
|
||||||
)
|
|
||||||
```
|
|
||||||
|
|
||||||
#### Pagination params dependency
|
|
||||||
|
|
||||||
!!! info "Added in `v2.4.1`"
|
|
||||||
|
|
||||||
Use [`offset_params()`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.offset_params) to generate a FastAPI dependency that injects `page` and `items_per_page` from query parameters with configurable defaults and a `max_page_size` cap:
|
|
||||||
|
|
||||||
```python
|
|
||||||
from typing import Annotated
|
|
||||||
from fastapi import Depends
|
|
||||||
|
|
||||||
@router.get("")
|
@router.get("")
|
||||||
async def list_users(
|
async def get_users(
|
||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
params: Annotated[dict, Depends(UserCrud.offset_params(default_page_size=20, max_page_size=100))],
|
params: Annotated[dict, Depends(UserCrud.offset_paginate_params(include_total=False))],
|
||||||
) -> OffsetPaginatedResponse[UserRead]:
|
) -> OffsetPaginatedResponse[UserRead]:
|
||||||
return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)
|
return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)
|
||||||
```
|
```
|
||||||
@@ -230,15 +208,9 @@ async def list_users(
|
|||||||
@router.get("")
|
@router.get("")
|
||||||
async def list_users(
|
async def list_users(
|
||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
cursor: str | None = None,
|
params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())],
|
||||||
items_per_page: int = 20,
|
|
||||||
) -> CursorPaginatedResponse[UserRead]:
|
) -> CursorPaginatedResponse[UserRead]:
|
||||||
return await UserCrud.cursor_paginate(
|
return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)
|
||||||
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 [`CursorPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPaginatedResponse):
|
The [`cursor_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_paginate) method returns a [`CursorPaginatedResponse`](../reference/schemas.md#fastapi_toolsets.schemas.CursorPaginatedResponse):
|
||||||
@@ -291,24 +263,6 @@ PostCrud = CrudFactory(model=Post, cursor_column=Post.id)
|
|||||||
PostCrud = CrudFactory(model=Post, cursor_column=Post.created_at)
|
PostCrud = CrudFactory(model=Post, cursor_column=Post.created_at)
|
||||||
```
|
```
|
||||||
|
|
||||||
#### Pagination params dependency
|
|
||||||
|
|
||||||
!!! info "Added in `v2.4.1`"
|
|
||||||
|
|
||||||
Use [`cursor_params()`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.cursor_params) to inject `cursor` and `items_per_page` from query parameters with a `max_page_size` cap:
|
|
||||||
|
|
||||||
```python
|
|
||||||
from typing import Annotated
|
|
||||||
from fastapi import Depends
|
|
||||||
|
|
||||||
@router.get("")
|
|
||||||
async def list_users(
|
|
||||||
session: SessionDep,
|
|
||||||
params: Annotated[dict, Depends(UserCrud.cursor_params(default_page_size=20, max_page_size=100))],
|
|
||||||
) -> CursorPaginatedResponse[UserRead]:
|
|
||||||
return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)
|
|
||||||
```
|
|
||||||
|
|
||||||
### Unified endpoint (both strategies)
|
### Unified endpoint (both strategies)
|
||||||
|
|
||||||
!!! info "Added in `v2.3.0`"
|
!!! info "Added in `v2.3.0`"
|
||||||
@@ -316,25 +270,14 @@ async def list_users(
|
|||||||
[`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.
|
[`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
|
```python
|
||||||
from fastapi_toolsets.crud import PaginationType
|
|
||||||
from fastapi_toolsets.schemas import PaginatedResponse
|
from fastapi_toolsets.schemas import PaginatedResponse
|
||||||
|
|
||||||
@router.get("")
|
@router.get("")
|
||||||
async def list_users(
|
async def list_users(
|
||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
pagination_type: PaginationType = PaginationType.OFFSET,
|
params: Annotated[dict, Depends(UserCrud.paginate_params())],
|
||||||
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]:
|
) -> PaginatedResponse[UserRead]:
|
||||||
return await UserCrud.paginate(
|
return await UserCrud.paginate(session, **params, schema=UserRead)
|
||||||
session,
|
|
||||||
pagination_type=pagination_type,
|
|
||||||
page=page,
|
|
||||||
cursor=cursor,
|
|
||||||
items_per_page=items_per_page,
|
|
||||||
schema=UserRead,
|
|
||||||
)
|
|
||||||
```
|
```
|
||||||
|
|
||||||
```
|
```
|
||||||
@@ -342,25 +285,6 @@ GET /users?pagination_type=offset&page=2&items_per_page=10
|
|||||||
GET /users?pagination_type=cursor&cursor=eyJ2YWx1ZSI6...&items_per_page=10
|
GET /users?pagination_type=cursor&cursor=eyJ2YWx1ZSI6...&items_per_page=10
|
||||||
```
|
```
|
||||||
|
|
||||||
#### Pagination params dependency
|
|
||||||
|
|
||||||
!!! info "Added in `v2.4.1`"
|
|
||||||
|
|
||||||
Use [`paginate_params()`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.paginate_params) to inject all parameters at once with configurable defaults and a `max_page_size` cap:
|
|
||||||
|
|
||||||
```python
|
|
||||||
from typing import Annotated
|
|
||||||
from fastapi import Depends
|
|
||||||
from fastapi_toolsets.schemas import PaginatedResponse
|
|
||||||
|
|
||||||
@router.get("")
|
|
||||||
async def list_users(
|
|
||||||
session: SessionDep,
|
|
||||||
params: Annotated[dict, Depends(UserCrud.paginate_params(default_page_size=20, max_page_size=100))],
|
|
||||||
) -> PaginatedResponse[UserRead]:
|
|
||||||
return await UserCrud.paginate(session, **params, schema=UserRead)
|
|
||||||
```
|
|
||||||
|
|
||||||
## Search
|
## 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).
|
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).
|
||||||
@@ -406,34 +330,18 @@ This allows searching with both [`offset_paginate`](../reference/crud.md#fastapi
|
|||||||
@router.get("")
|
@router.get("")
|
||||||
async def get_users(
|
async def get_users(
|
||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
items_per_page: int = 50,
|
params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
|
||||||
page: int = 1,
|
|
||||||
search: str | None = None,
|
|
||||||
) -> OffsetPaginatedResponse[UserRead]:
|
) -> OffsetPaginatedResponse[UserRead]:
|
||||||
return await UserCrud.offset_paginate(
|
return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)
|
||||||
session=session,
|
|
||||||
items_per_page=items_per_page,
|
|
||||||
page=page,
|
|
||||||
search=search,
|
|
||||||
schema=UserRead,
|
|
||||||
)
|
|
||||||
```
|
```
|
||||||
|
|
||||||
```python
|
```python
|
||||||
@router.get("")
|
@router.get("")
|
||||||
async def get_users(
|
async def get_users(
|
||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
cursor: str | None = None,
|
params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())],
|
||||||
items_per_page: int = 50,
|
|
||||||
search: str | None = None,
|
|
||||||
) -> CursorPaginatedResponse[UserRead]:
|
) -> CursorPaginatedResponse[UserRead]:
|
||||||
return await UserCrud.cursor_paginate(
|
return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead)
|
||||||
session=session,
|
|
||||||
items_per_page=items_per_page,
|
|
||||||
cursor=cursor,
|
|
||||||
search=search,
|
|
||||||
schema=UserRead,
|
|
||||||
)
|
|
||||||
```
|
```
|
||||||
|
|
||||||
### Faceted search
|
### Faceted search
|
||||||
@@ -486,7 +394,7 @@ Use `filter_by` to pass the client's chosen filter values directly — no need t
|
|||||||
|
|
||||||
`filter_by` and `filters` can be combined — both are applied with AND logic.
|
`filter_by` and `filters` can be combined — both are applied with AND logic.
|
||||||
|
|
||||||
Use [`filter_params()`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.filter_params) to generate a dict with the facet filter values from the query parameters:
|
Facet filtering is built into the consolidated params dependencies. When `filter=True` (the default), facet fields are exposed as query parameters and collected into `filter_by` automatically:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
from typing import Annotated
|
from typing import Annotated
|
||||||
@@ -501,13 +409,11 @@ UserCrud = CrudFactory(
|
|||||||
@router.get("", response_model_exclude_none=True)
|
@router.get("", response_model_exclude_none=True)
|
||||||
async def list_users(
|
async def list_users(
|
||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
page: int = 1,
|
params: Annotated[dict, Depends(UserCrud.offset_paginate_params())],
|
||||||
filter_by: Annotated[dict[str, list[str]], Depends(UserCrud.filter_params())],
|
|
||||||
) -> OffsetPaginatedResponse[UserRead]:
|
) -> OffsetPaginatedResponse[UserRead]:
|
||||||
return await UserCrud.offset_paginate(
|
return await UserCrud.offset_paginate(
|
||||||
session=session,
|
session=session,
|
||||||
page=page,
|
**params,
|
||||||
filter_by=filter_by,
|
|
||||||
schema=UserRead,
|
schema=UserRead,
|
||||||
)
|
)
|
||||||
```
|
```
|
||||||
@@ -536,20 +442,21 @@ UserCrud = CrudFactory(
|
|||||||
)
|
)
|
||||||
```
|
```
|
||||||
|
|
||||||
Call [`order_params()`](../reference/crud.md#fastapi_toolsets.crud.factory.AsyncCrud.order_params) to generate a FastAPI dependency that maps the query parameters to an [`OrderByClause`](../reference/crud.md#fastapi_toolsets.crud.factory.OrderByClause) expression:
|
Ordering is built into the consolidated params dependencies. When `order=True` (the default), `order_by` and `order` query parameters are exposed and resolved into an `OrderByClause` automatically:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
from typing import Annotated
|
from typing import Annotated
|
||||||
|
|
||||||
from fastapi import Depends
|
from fastapi import Depends
|
||||||
from fastapi_toolsets.crud import OrderByClause
|
|
||||||
|
|
||||||
@router.get("")
|
@router.get("")
|
||||||
async def list_users(
|
async def list_users(
|
||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
order_by: Annotated[OrderByClause | None, Depends(UserCrud.order_params())],
|
params: Annotated[dict, Depends(UserCrud.offset_paginate_params(
|
||||||
|
default_order_field=User.created_at,
|
||||||
|
))],
|
||||||
) -> OffsetPaginatedResponse[UserRead]:
|
) -> OffsetPaginatedResponse[UserRead]:
|
||||||
return await UserCrud.offset_paginate(session=session, order_by=order_by, schema=UserRead)
|
return await UserCrud.offset_paginate(session=session, **params, schema=UserRead)
|
||||||
```
|
```
|
||||||
|
|
||||||
The dependency adds two query parameters to the endpoint:
|
The dependency adds two query parameters to the endpoint:
|
||||||
@@ -566,10 +473,10 @@ GET /users?order_by=name&order=desc → ORDER BY users.name DESC
|
|||||||
|
|
||||||
An unknown `order_by` value raises [`InvalidOrderFieldError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidOrderFieldError) (HTTP 422).
|
An unknown `order_by` value raises [`InvalidOrderFieldError`](../reference/exceptions.md#fastapi_toolsets.exceptions.exceptions.InvalidOrderFieldError) (HTTP 422).
|
||||||
|
|
||||||
You can also pass `order_fields` directly to `order_params()` to override the class-level defaults without modifying them:
|
You can also pass `order_fields` directly to override the class-level defaults:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
UserOrderParams = UserCrud.order_params(order_fields=[User.name])
|
params = UserCrud.offset_paginate_params(order_fields=[User.name])
|
||||||
```
|
```
|
||||||
|
|
||||||
## Relationship loading
|
## Relationship loading
|
||||||
@@ -656,12 +563,11 @@ async def get_user(session: SessionDep, uuid: UUID) -> Response[UserRead]:
|
|||||||
)
|
)
|
||||||
|
|
||||||
@router.get("")
|
@router.get("")
|
||||||
async def list_users(session: SessionDep, page: int = 1) -> OffsetPaginatedResponse[UserRead]:
|
async def list_users(
|
||||||
return await crud.UserCrud.offset_paginate(
|
session: SessionDep,
|
||||||
session=session,
|
params: Annotated[dict, Depends(crud.UserCrud.offset_paginate_params())],
|
||||||
page=page,
|
) -> OffsetPaginatedResponse[UserRead]:
|
||||||
schema=UserRead,
|
return await crud.UserCrud.offset_paginate(session=session, **params, schema=UserRead)
|
||||||
)
|
|
||||||
```
|
```
|
||||||
|
|
||||||
The schema must have `from_attributes=True` (or inherit from [`PydanticBase`](../reference/schemas.md#fastapi_toolsets.schemas.PydanticBase)) so it can be built from SQLAlchemy model instances.
|
The schema must have `from_attributes=True` (or inherit from [`PydanticBase`](../reference/schemas.md#fastapi_toolsets.schemas.PydanticBase)) so it can be built from SQLAlchemy model instances.
|
||||||
|
|||||||
@@ -79,9 +79,6 @@ The examples above are already compatible with parallel test execution with `pyt
|
|||||||
|
|
||||||
## Cleaning up tables
|
## Cleaning up tables
|
||||||
|
|
||||||
!!! warning
|
|
||||||
Since `V2.1.0` `cleanup_tables` now live in `fastapi_toolsets.db`. For backward compatibility the function is still available in `fastapi_toolsets.pytest`, but this will be remove in `V3.0.0`.
|
|
||||||
|
|
||||||
If you want to manually clean up a database you can use [`cleanup_tables`](../reference/db.md#fastapi_toolsets.db.cleanup_tables), this will truncate all tables between tests for fast isolation:
|
If you want to manually clean up a database you can use [`cleanup_tables`](../reference/db.md#fastapi_toolsets.db.cleanup_tables), this will truncate all tables between tests for fast isolation:
|
||||||
|
|
||||||
```python
|
```python
|
||||||
|
|||||||
@@ -2,7 +2,6 @@ from typing import Annotated
|
|||||||
|
|
||||||
from fastapi import APIRouter, Depends
|
from fastapi import APIRouter, Depends
|
||||||
|
|
||||||
from fastapi_toolsets.crud import OrderByClause
|
|
||||||
from fastapi_toolsets.schemas import (
|
from fastapi_toolsets.schemas import (
|
||||||
CursorPaginatedResponse,
|
CursorPaginatedResponse,
|
||||||
OffsetPaginatedResponse,
|
OffsetPaginatedResponse,
|
||||||
@@ -22,21 +21,18 @@ async def list_articles_offset(
|
|||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
params: Annotated[
|
params: Annotated[
|
||||||
dict,
|
dict,
|
||||||
Depends(ArticleCrud.offset_params(default_page_size=20, max_page_size=100)),
|
Depends(
|
||||||
|
ArticleCrud.offset_paginate_params(
|
||||||
|
default_page_size=20,
|
||||||
|
max_page_size=100,
|
||||||
|
default_order_field=Article.created_at,
|
||||||
|
)
|
||||||
|
),
|
||||||
],
|
],
|
||||||
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)),
|
|
||||||
],
|
|
||||||
search: str | None = None,
|
|
||||||
) -> OffsetPaginatedResponse[ArticleRead]:
|
) -> OffsetPaginatedResponse[ArticleRead]:
|
||||||
return await ArticleCrud.offset_paginate(
|
return await ArticleCrud.offset_paginate(
|
||||||
session=session,
|
session=session,
|
||||||
**params,
|
**params,
|
||||||
search=search,
|
|
||||||
filter_by=filter_by or None,
|
|
||||||
order_by=order_by,
|
|
||||||
schema=ArticleRead,
|
schema=ArticleRead,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -46,21 +42,18 @@ async def list_articles_cursor(
|
|||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
params: Annotated[
|
params: Annotated[
|
||||||
dict,
|
dict,
|
||||||
Depends(ArticleCrud.cursor_params(default_page_size=20, max_page_size=100)),
|
Depends(
|
||||||
|
ArticleCrud.cursor_paginate_params(
|
||||||
|
default_page_size=20,
|
||||||
|
max_page_size=100,
|
||||||
|
default_order_field=Article.created_at,
|
||||||
|
)
|
||||||
|
),
|
||||||
],
|
],
|
||||||
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)),
|
|
||||||
],
|
|
||||||
search: str | None = None,
|
|
||||||
) -> CursorPaginatedResponse[ArticleRead]:
|
) -> CursorPaginatedResponse[ArticleRead]:
|
||||||
return await ArticleCrud.cursor_paginate(
|
return await ArticleCrud.cursor_paginate(
|
||||||
session=session,
|
session=session,
|
||||||
**params,
|
**params,
|
||||||
search=search,
|
|
||||||
filter_by=filter_by or None,
|
|
||||||
order_by=order_by,
|
|
||||||
schema=ArticleRead,
|
schema=ArticleRead,
|
||||||
)
|
)
|
||||||
|
|
||||||
@@ -70,20 +63,17 @@ async def list_articles(
|
|||||||
session: SessionDep,
|
session: SessionDep,
|
||||||
params: Annotated[
|
params: Annotated[
|
||||||
dict,
|
dict,
|
||||||
Depends(ArticleCrud.paginate_params(default_page_size=20, max_page_size=100)),
|
Depends(
|
||||||
|
ArticleCrud.paginate_params(
|
||||||
|
default_page_size=20,
|
||||||
|
max_page_size=100,
|
||||||
|
default_order_field=Article.created_at,
|
||||||
|
)
|
||||||
|
),
|
||||||
],
|
],
|
||||||
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)),
|
|
||||||
],
|
|
||||||
search: str | None = None,
|
|
||||||
) -> PaginatedResponse[ArticleRead]:
|
) -> PaginatedResponse[ArticleRead]:
|
||||||
return await ArticleCrud.paginate(
|
return await ArticleCrud.paginate(
|
||||||
session,
|
session,
|
||||||
**params,
|
**params,
|
||||||
search=search,
|
|
||||||
filter_by=filter_by or None,
|
|
||||||
order_by=order_by,
|
|
||||||
schema=ArticleRead,
|
schema=ArticleRead,
|
||||||
)
|
)
|
||||||
|
|||||||
@@ -2,6 +2,7 @@
|
|||||||
|
|
||||||
from ..exceptions import (
|
from ..exceptions import (
|
||||||
InvalidFacetFilterError,
|
InvalidFacetFilterError,
|
||||||
|
InvalidSearchColumnError,
|
||||||
NoSearchableFieldsError,
|
NoSearchableFieldsError,
|
||||||
UnsupportedFacetTypeError,
|
UnsupportedFacetTypeError,
|
||||||
)
|
)
|
||||||
@@ -22,6 +23,7 @@ __all__ = [
|
|||||||
"FacetFieldType",
|
"FacetFieldType",
|
||||||
"get_searchable_fields",
|
"get_searchable_fields",
|
||||||
"InvalidFacetFilterError",
|
"InvalidFacetFilterError",
|
||||||
|
"InvalidSearchColumnError",
|
||||||
"JoinType",
|
"JoinType",
|
||||||
"M2MFieldType",
|
"M2MFieldType",
|
||||||
"NoSearchableFieldsError",
|
"NoSearchableFieldsError",
|
||||||
|
|||||||
@@ -47,6 +47,7 @@ from .search import (
|
|||||||
build_filter_by,
|
build_filter_by,
|
||||||
build_search_filters,
|
build_search_filters,
|
||||||
facet_keys,
|
facet_keys,
|
||||||
|
search_field_keys,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -263,118 +264,285 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
)
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def filter_params(
|
def _resolve_search_columns(
|
||||||
|
cls: type[Self],
|
||||||
|
search_fields: Sequence[SearchFieldType] | None,
|
||||||
|
) -> list[str] | None:
|
||||||
|
"""Return search column keys, or None if no searchable fields configured."""
|
||||||
|
fields = search_fields if search_fields is not None else cls.searchable_fields
|
||||||
|
if not fields:
|
||||||
|
return None
|
||||||
|
return search_field_keys(fields)
|
||||||
|
|
||||||
|
@classmethod
|
||||||
|
def _build_paginate_params(
|
||||||
cls: type[Self],
|
cls: type[Self],
|
||||||
*,
|
*,
|
||||||
facet_fields: Sequence[FacetFieldType] | None = None,
|
pagination_params: list[inspect.Parameter],
|
||||||
) -> Callable[..., Awaitable[dict[str, list[str]]]]:
|
pagination_fixed: dict[str, Any],
|
||||||
"""Return a FastAPI dependency that collects facet filter values from query parameters.
|
dep_name: str,
|
||||||
|
search: bool,
|
||||||
|
filter: bool,
|
||||||
|
order: bool,
|
||||||
|
search_fields: Sequence[SearchFieldType] | None,
|
||||||
|
facet_fields: Sequence[FacetFieldType] | None,
|
||||||
|
order_fields: Sequence[QueryableAttribute[Any]] | None,
|
||||||
|
default_order_field: QueryableAttribute[Any] | None,
|
||||||
|
default_order: Literal["asc", "desc"],
|
||||||
|
) -> Callable[..., Awaitable[dict[str, Any]]]:
|
||||||
|
"""Build a consolidated FastAPI dependency that merges pagination, search, filter, and order params."""
|
||||||
|
all_params: list[inspect.Parameter] = list(pagination_params)
|
||||||
|
pagination_param_names = tuple(p.name for p in pagination_params)
|
||||||
|
reserved_names: set[str] = set(pagination_param_names)
|
||||||
|
|
||||||
Args:
|
search_keys: list[str] | None = None
|
||||||
facet_fields: Override the facet fields for this dependency. Falls back to the
|
if search:
|
||||||
class-level ``facet_fields`` if not provided.
|
search_keys = cls._resolve_search_columns(search_fields)
|
||||||
|
if search_keys:
|
||||||
Returns:
|
all_params.extend(
|
||||||
An async dependency function named ``{Model}FilterParams`` that resolves to a
|
[
|
||||||
``dict[str, list[str]]`` containing only the keys that were supplied in the
|
inspect.Parameter(
|
||||||
request (absent/``None`` parameters are excluded).
|
"search",
|
||||||
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
Raises:
|
annotation=str | None,
|
||||||
ValueError: If no facet fields are configured on this CRUD class and none are
|
default=Query(
|
||||||
provided via ``facet_fields``.
|
default=None, description="Search query string"
|
||||||
"""
|
),
|
||||||
fields = cls._resolve_facet_fields(facet_fields)
|
),
|
||||||
if not fields:
|
inspect.Parameter(
|
||||||
raise ValueError(
|
"search_column",
|
||||||
f"{cls.__name__} has no facet_fields configured. "
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
"Pass facet_fields= or set them on CrudFactory."
|
annotation=str | None,
|
||||||
|
default=Query(
|
||||||
|
default=None,
|
||||||
|
description="Restrict search to a single column",
|
||||||
|
enum=search_keys,
|
||||||
|
),
|
||||||
|
),
|
||||||
|
]
|
||||||
)
|
)
|
||||||
keys = facet_keys(fields)
|
reserved_names.update({"search", "search_column"})
|
||||||
|
|
||||||
async def dependency(**kwargs: Any) -> dict[str, list[str]]:
|
filter_keys: list[str] | None = None
|
||||||
return {k: v for k, v in kwargs.items() if v is not None}
|
if filter:
|
||||||
|
resolved_facets = cls._resolve_facet_fields(facet_fields)
|
||||||
dependency.__name__ = f"{cls.model.__name__}FilterParams"
|
if resolved_facets:
|
||||||
dependency.__signature__ = inspect.Signature( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
|
filter_keys = facet_keys(resolved_facets)
|
||||||
parameters=[
|
for k in filter_keys:
|
||||||
|
if k in reserved_names:
|
||||||
|
raise ValueError(
|
||||||
|
f"Facet field key {k!r} conflicts with a reserved "
|
||||||
|
f"parameter name. Reserved names: {sorted(reserved_names)}"
|
||||||
|
)
|
||||||
|
all_params.extend(
|
||||||
inspect.Parameter(
|
inspect.Parameter(
|
||||||
k,
|
k,
|
||||||
inspect.Parameter.KEYWORD_ONLY,
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
annotation=list[str] | None,
|
annotation=list[str] | None,
|
||||||
default=Query(default=None),
|
default=Query(default=None),
|
||||||
)
|
)
|
||||||
for k in keys
|
for k in filter_keys
|
||||||
|
)
|
||||||
|
reserved_names.update(filter_keys)
|
||||||
|
|
||||||
|
order_field_map: dict[str, QueryableAttribute[Any]] | None = None
|
||||||
|
order_valid_keys: list[str] | None = None
|
||||||
|
if order:
|
||||||
|
resolved_order = (
|
||||||
|
order_fields if order_fields is not None else cls.order_fields
|
||||||
|
)
|
||||||
|
if resolved_order:
|
||||||
|
order_field_map = {f.key: f for f in resolved_order}
|
||||||
|
order_valid_keys = sorted(order_field_map.keys())
|
||||||
|
all_params.extend(
|
||||||
|
[
|
||||||
|
inspect.Parameter(
|
||||||
|
"order_by",
|
||||||
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
|
annotation=str | None,
|
||||||
|
default=Query(
|
||||||
|
None,
|
||||||
|
description=f"Field to order by. Valid values: {order_valid_keys}",
|
||||||
|
enum=order_valid_keys,
|
||||||
|
),
|
||||||
|
),
|
||||||
|
inspect.Parameter(
|
||||||
|
"order",
|
||||||
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
|
annotation=Literal["asc", "desc"],
|
||||||
|
default=Query(default_order, description="Sort direction"),
|
||||||
|
),
|
||||||
]
|
]
|
||||||
)
|
)
|
||||||
|
|
||||||
|
async def dependency(**kwargs: Any) -> dict[str, Any]:
|
||||||
|
result: dict[str, Any] = dict(pagination_fixed)
|
||||||
|
for name in pagination_param_names:
|
||||||
|
result[name] = kwargs[name]
|
||||||
|
|
||||||
|
if search_keys is not None:
|
||||||
|
search_val = kwargs.get("search")
|
||||||
|
if search_val is not None:
|
||||||
|
result["search"] = search_val
|
||||||
|
search_col_val = kwargs.get("search_column")
|
||||||
|
if search_col_val is not None:
|
||||||
|
result["search_column"] = search_col_val
|
||||||
|
|
||||||
|
if filter_keys is not None:
|
||||||
|
filter_by = {
|
||||||
|
k: kwargs[k] for k in filter_keys if kwargs.get(k) is not None
|
||||||
|
}
|
||||||
|
result["filter_by"] = filter_by or None
|
||||||
|
|
||||||
|
if order_field_map is not None:
|
||||||
|
order_by_val = kwargs.get("order_by")
|
||||||
|
order_dir = kwargs.get("order", default_order)
|
||||||
|
if order_by_val is None:
|
||||||
|
field = default_order_field
|
||||||
|
elif order_by_val not in order_field_map:
|
||||||
|
raise InvalidOrderFieldError(order_by_val, order_valid_keys or [])
|
||||||
|
else:
|
||||||
|
field = order_field_map[order_by_val]
|
||||||
|
if field is not None:
|
||||||
|
result["order_by"] = (
|
||||||
|
field.asc() if order_dir == "asc" else field.desc()
|
||||||
|
)
|
||||||
|
else:
|
||||||
|
result["order_by"] = None
|
||||||
|
|
||||||
|
return result
|
||||||
|
|
||||||
|
dependency.__name__ = dep_name
|
||||||
|
dependency.__signature__ = inspect.Signature( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute]
|
||||||
|
parameters=all_params,
|
||||||
|
)
|
||||||
return dependency
|
return dependency
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def offset_params(
|
def offset_paginate_params(
|
||||||
cls: type[Self],
|
cls: type[Self],
|
||||||
*,
|
*,
|
||||||
default_page_size: int = 20,
|
default_page_size: int = 20,
|
||||||
max_page_size: int = 100,
|
max_page_size: int = 100,
|
||||||
include_total: bool = True,
|
include_total: bool = True,
|
||||||
|
search: bool = True,
|
||||||
|
filter: bool = True,
|
||||||
|
order: bool = True,
|
||||||
|
search_fields: Sequence[SearchFieldType] | None = None,
|
||||||
|
facet_fields: Sequence[FacetFieldType] | None = None,
|
||||||
|
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
|
||||||
|
default_order_field: QueryableAttribute[Any] | None = None,
|
||||||
|
default_order: Literal["asc", "desc"] = "asc",
|
||||||
) -> Callable[..., Awaitable[dict[str, Any]]]:
|
) -> Callable[..., Awaitable[dict[str, Any]]]:
|
||||||
"""Return a FastAPI dependency that collects offset pagination params from query params.
|
"""Return a FastAPI dependency that collects all params for :meth:`offset_paginate`.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
default_page_size: Default value for the ``items_per_page`` query parameter.
|
default_page_size: Default ``items_per_page`` value.
|
||||||
max_page_size: Maximum allowed value for ``items_per_page`` (enforced via
|
max_page_size: Maximum ``items_per_page`` value.
|
||||||
``le`` on the ``Query``).
|
include_total: Whether to include total count (not a query param).
|
||||||
include_total: Server-side flag forwarded as-is to ``include_total`` in
|
search: Enable search query parameters.
|
||||||
:meth:`offset_paginate`. Not exposed as a query parameter.
|
filter: Enable facet filter query parameters.
|
||||||
|
order: Enable order query parameters.
|
||||||
|
search_fields: Override searchable fields.
|
||||||
|
facet_fields: Override facet fields.
|
||||||
|
order_fields: Override order fields.
|
||||||
|
default_order_field: Default field to order by when ``order_by`` is absent.
|
||||||
|
default_order: Default sort direction.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
An async dependency that resolves to a dict with ``page``,
|
An async dependency that resolves to a dict ready to be unpacked
|
||||||
``items_per_page``, and ``include_total`` keys, ready to be
|
into :meth:`offset_paginate`.
|
||||||
unpacked into :meth:`offset_paginate`.
|
|
||||||
"""
|
"""
|
||||||
|
pagination_params = [
|
||||||
async def dependency(
|
inspect.Parameter(
|
||||||
page: int = Query(1, ge=1, description="Page number (1-indexed)"),
|
"page",
|
||||||
items_per_page: int = _page_size_query(default_page_size, max_page_size),
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
) -> dict[str, Any]:
|
annotation=int,
|
||||||
return {
|
default=Query(1, ge=1, description="Page number (1-indexed)"),
|
||||||
"page": page,
|
),
|
||||||
"items_per_page": items_per_page,
|
inspect.Parameter(
|
||||||
"include_total": include_total,
|
"items_per_page",
|
||||||
}
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
|
annotation=int,
|
||||||
dependency.__name__ = f"{cls.model.__name__}OffsetParams"
|
default=_page_size_query(default_page_size, max_page_size),
|
||||||
return dependency
|
),
|
||||||
|
]
|
||||||
|
return cls._build_paginate_params(
|
||||||
|
pagination_params=pagination_params,
|
||||||
|
pagination_fixed={"include_total": include_total},
|
||||||
|
dep_name=f"{cls.model.__name__}OffsetPaginateParams",
|
||||||
|
search=search,
|
||||||
|
filter=filter,
|
||||||
|
order=order,
|
||||||
|
search_fields=search_fields,
|
||||||
|
facet_fields=facet_fields,
|
||||||
|
order_fields=order_fields,
|
||||||
|
default_order_field=default_order_field,
|
||||||
|
default_order=default_order,
|
||||||
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def cursor_params(
|
def cursor_paginate_params(
|
||||||
cls: type[Self],
|
cls: type[Self],
|
||||||
*,
|
*,
|
||||||
default_page_size: int = 20,
|
default_page_size: int = 20,
|
||||||
max_page_size: int = 100,
|
max_page_size: int = 100,
|
||||||
|
search: bool = True,
|
||||||
|
filter: bool = True,
|
||||||
|
order: bool = True,
|
||||||
|
search_fields: Sequence[SearchFieldType] | None = None,
|
||||||
|
facet_fields: Sequence[FacetFieldType] | None = None,
|
||||||
|
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
|
||||||
|
default_order_field: QueryableAttribute[Any] | None = None,
|
||||||
|
default_order: Literal["asc", "desc"] = "asc",
|
||||||
) -> Callable[..., Awaitable[dict[str, Any]]]:
|
) -> Callable[..., Awaitable[dict[str, Any]]]:
|
||||||
"""Return a FastAPI dependency that collects cursor pagination params from query params.
|
"""Return a FastAPI dependency that collects all params for :meth:`cursor_paginate`.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
default_page_size: Default value for the ``items_per_page`` query parameter.
|
default_page_size: Default ``items_per_page`` value.
|
||||||
max_page_size: Maximum allowed value for ``items_per_page`` (enforced via
|
max_page_size: Maximum ``items_per_page`` value.
|
||||||
``le`` on the ``Query``).
|
search: Enable search query parameters.
|
||||||
|
filter: Enable facet filter query parameters.
|
||||||
|
order: Enable order query parameters.
|
||||||
|
search_fields: Override searchable fields.
|
||||||
|
facet_fields: Override facet fields.
|
||||||
|
order_fields: Override order fields.
|
||||||
|
default_order_field: Default field to order by when ``order_by`` is absent.
|
||||||
|
default_order: Default sort direction.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
An async dependency that resolves to a dict with ``cursor`` and
|
An async dependency that resolves to a dict ready to be unpacked
|
||||||
``items_per_page`` keys, ready to be unpacked into
|
into :meth:`cursor_paginate`.
|
||||||
:meth:`cursor_paginate`.
|
|
||||||
"""
|
"""
|
||||||
|
pagination_params = [
|
||||||
async def dependency(
|
inspect.Parameter(
|
||||||
cursor: str | None = Query(
|
"cursor",
|
||||||
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
|
annotation=str | None,
|
||||||
|
default=Query(
|
||||||
None, description="Cursor token from a previous response"
|
None, description="Cursor token from a previous response"
|
||||||
),
|
),
|
||||||
items_per_page: int = _page_size_query(default_page_size, max_page_size),
|
),
|
||||||
) -> dict[str, Any]:
|
inspect.Parameter(
|
||||||
return {"cursor": cursor, "items_per_page": items_per_page}
|
"items_per_page",
|
||||||
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
dependency.__name__ = f"{cls.model.__name__}CursorParams"
|
annotation=int,
|
||||||
return dependency
|
default=_page_size_query(default_page_size, max_page_size),
|
||||||
|
),
|
||||||
|
]
|
||||||
|
return cls._build_paginate_params(
|
||||||
|
pagination_params=pagination_params,
|
||||||
|
pagination_fixed={},
|
||||||
|
dep_name=f"{cls.model.__name__}CursorPaginateParams",
|
||||||
|
search=search,
|
||||||
|
filter=filter,
|
||||||
|
order=order,
|
||||||
|
search_fields=search_fields,
|
||||||
|
facet_fields=facet_fields,
|
||||||
|
order_fields=order_fields,
|
||||||
|
default_order_field=default_order_field,
|
||||||
|
default_order=default_order,
|
||||||
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
def paginate_params(
|
def paginate_params(
|
||||||
@@ -384,102 +552,81 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
max_page_size: int = 100,
|
max_page_size: int = 100,
|
||||||
default_pagination_type: PaginationType = PaginationType.OFFSET,
|
default_pagination_type: PaginationType = PaginationType.OFFSET,
|
||||||
include_total: bool = True,
|
include_total: bool = True,
|
||||||
|
search: bool = True,
|
||||||
|
filter: bool = True,
|
||||||
|
order: bool = True,
|
||||||
|
search_fields: Sequence[SearchFieldType] | None = None,
|
||||||
|
facet_fields: Sequence[FacetFieldType] | None = None,
|
||||||
|
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
|
||||||
|
default_order_field: QueryableAttribute[Any] | None = None,
|
||||||
|
default_order: Literal["asc", "desc"] = "asc",
|
||||||
) -> Callable[..., Awaitable[dict[str, Any]]]:
|
) -> Callable[..., Awaitable[dict[str, Any]]]:
|
||||||
"""Return a FastAPI dependency that collects all pagination params from query params.
|
"""Return a FastAPI dependency that collects all params for :meth:`paginate`.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
default_page_size: Default value for the ``items_per_page`` query parameter.
|
default_page_size: Default ``items_per_page`` value.
|
||||||
max_page_size: Maximum allowed value for ``items_per_page`` (enforced via
|
max_page_size: Maximum ``items_per_page`` value.
|
||||||
``le`` on the ``Query``).
|
|
||||||
default_pagination_type: Default pagination strategy.
|
default_pagination_type: Default pagination strategy.
|
||||||
include_total: Server-side flag forwarded as-is to ``include_total`` in
|
include_total: Whether to include total count (not a query param).
|
||||||
:meth:`paginate`. Not exposed as a query parameter.
|
search: Enable search query parameters.
|
||||||
|
filter: Enable facet filter query parameters.
|
||||||
|
order: Enable order query parameters.
|
||||||
|
search_fields: Override searchable fields.
|
||||||
|
facet_fields: Override facet fields.
|
||||||
|
order_fields: Override order fields.
|
||||||
|
default_order_field: Default field to order by when ``order_by`` is absent.
|
||||||
|
default_order: Default sort direction.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
An async dependency that resolves to a dict with ``pagination_type``,
|
An async dependency that resolves to a dict ready to be unpacked
|
||||||
``page``, ``cursor``, ``items_per_page``, and ``include_total`` keys,
|
into :meth:`paginate`.
|
||||||
ready to be unpacked into :meth:`paginate`.
|
|
||||||
"""
|
"""
|
||||||
|
pagination_params = [
|
||||||
async def dependency(
|
inspect.Parameter(
|
||||||
pagination_type: PaginationType = Query(
|
"pagination_type",
|
||||||
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
|
annotation=PaginationType,
|
||||||
|
default=Query(
|
||||||
default_pagination_type, description="Pagination strategy"
|
default_pagination_type, description="Pagination strategy"
|
||||||
),
|
),
|
||||||
page: int = Query(
|
),
|
||||||
|
inspect.Parameter(
|
||||||
|
"page",
|
||||||
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
|
annotation=int,
|
||||||
|
default=Query(
|
||||||
1, ge=1, description="Page number (1-indexed, offset only)"
|
1, ge=1, description="Page number (1-indexed, offset only)"
|
||||||
),
|
),
|
||||||
cursor: str | None = Query(
|
|
||||||
None, description="Cursor token from a previous response (cursor only)"
|
|
||||||
),
|
),
|
||||||
items_per_page: int = _page_size_query(default_page_size, max_page_size),
|
inspect.Parameter(
|
||||||
) -> dict[str, Any]:
|
"cursor",
|
||||||
return {
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
"pagination_type": pagination_type,
|
annotation=str | None,
|
||||||
"page": page,
|
default=Query(
|
||||||
"cursor": cursor,
|
None,
|
||||||
"items_per_page": items_per_page,
|
description="Cursor token from a previous response (cursor only)",
|
||||||
"include_total": include_total,
|
),
|
||||||
}
|
),
|
||||||
|
inspect.Parameter(
|
||||||
dependency.__name__ = f"{cls.model.__name__}PaginateParams"
|
"items_per_page",
|
||||||
return dependency
|
inspect.Parameter.KEYWORD_ONLY,
|
||||||
|
annotation=int,
|
||||||
@classmethod
|
default=_page_size_query(default_page_size, max_page_size),
|
||||||
def order_params(
|
),
|
||||||
cls: type[Self],
|
]
|
||||||
*,
|
return cls._build_paginate_params(
|
||||||
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
|
pagination_params=pagination_params,
|
||||||
default_field: QueryableAttribute[Any] | None = None,
|
pagination_fixed={"include_total": include_total},
|
||||||
default_order: Literal["asc", "desc"] = "asc",
|
dep_name=f"{cls.model.__name__}PaginateParams",
|
||||||
) -> Callable[..., Awaitable[OrderByClause | None]]:
|
search=search,
|
||||||
"""Return a FastAPI dependency that resolves order query params into an order_by clause.
|
filter=filter,
|
||||||
|
order=order,
|
||||||
Args:
|
search_fields=search_fields,
|
||||||
order_fields: Override the allowed order fields. Falls back to the class-level
|
facet_fields=facet_fields,
|
||||||
``order_fields`` if not provided.
|
order_fields=order_fields,
|
||||||
default_field: Field to order by when ``order_by`` query param is absent.
|
default_order_field=default_order_field,
|
||||||
If ``None`` and no ``order_by`` is provided, no ordering is applied.
|
default_order=default_order,
|
||||||
default_order: Default order direction when ``order`` is absent
|
|
||||||
(``"asc"`` or ``"desc"``).
|
|
||||||
|
|
||||||
Returns:
|
|
||||||
An async dependency function named ``{Model}OrderParams`` that resolves to an
|
|
||||||
``OrderByClause`` (or ``None``). Pass it to ``Depends()`` in your route.
|
|
||||||
|
|
||||||
Raises:
|
|
||||||
ValueError: If no order fields are configured on this CRUD class and none are
|
|
||||||
provided via ``order_fields``.
|
|
||||||
InvalidOrderFieldError: When the request provides an unknown ``order_by`` value.
|
|
||||||
"""
|
|
||||||
fields = order_fields if order_fields is not None else cls.order_fields
|
|
||||||
if not fields:
|
|
||||||
raise ValueError(
|
|
||||||
f"{cls.__name__} has no order_fields configured. "
|
|
||||||
"Pass order_fields= or set them on CrudFactory."
|
|
||||||
)
|
)
|
||||||
field_map: dict[str, QueryableAttribute[Any]] = {f.key: f for f in fields}
|
|
||||||
valid_keys = sorted(field_map.keys())
|
|
||||||
|
|
||||||
async def dependency(
|
|
||||||
order_by: str | None = Query(
|
|
||||||
None, description=f"Field to order by. Valid values: {valid_keys}"
|
|
||||||
),
|
|
||||||
order: Literal["asc", "desc"] = Query(
|
|
||||||
default_order, description="Sort direction"
|
|
||||||
),
|
|
||||||
) -> OrderByClause | None:
|
|
||||||
if order_by is None:
|
|
||||||
if default_field is None:
|
|
||||||
return None
|
|
||||||
field = default_field
|
|
||||||
elif order_by not in field_map:
|
|
||||||
raise InvalidOrderFieldError(order_by, valid_keys)
|
|
||||||
else:
|
|
||||||
field = field_map[order_by]
|
|
||||||
return field.asc() if order == "asc" else field.desc()
|
|
||||||
|
|
||||||
dependency.__name__ = f"{cls.model.__name__}OrderParams"
|
|
||||||
return dependency
|
|
||||||
|
|
||||||
@overload
|
@overload
|
||||||
@classmethod
|
@classmethod
|
||||||
@@ -1056,6 +1203,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
include_total: bool = True,
|
include_total: bool = True,
|
||||||
search: str | SearchConfig | None = None,
|
search: str | SearchConfig | None = None,
|
||||||
search_fields: Sequence[SearchFieldType] | None = None,
|
search_fields: Sequence[SearchFieldType] | None = None,
|
||||||
|
search_column: str | None = None,
|
||||||
facet_fields: Sequence[FacetFieldType] | None = None,
|
facet_fields: Sequence[FacetFieldType] | None = None,
|
||||||
filter_by: dict[str, Any] | BaseModel | None = None,
|
filter_by: dict[str, Any] | BaseModel | None = None,
|
||||||
schema: type[BaseModel],
|
schema: type[BaseModel],
|
||||||
@@ -1075,6 +1223,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
``pagination.total_count`` will be ``None``.
|
``pagination.total_count`` will be ``None``.
|
||||||
search: Search query string or SearchConfig object
|
search: Search query string or SearchConfig object
|
||||||
search_fields: Fields to search in (overrides class default)
|
search_fields: Fields to search in (overrides class default)
|
||||||
|
search_column: Restrict search to a single column key.
|
||||||
facet_fields: Columns to compute distinct values for (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.
|
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,
|
Keys must match the column.key of a facet field. Scalar → equality,
|
||||||
@@ -1097,6 +1246,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
search,
|
search,
|
||||||
search_fields=search_fields,
|
search_fields=search_fields,
|
||||||
default_fields=cls.searchable_fields,
|
default_fields=cls.searchable_fields,
|
||||||
|
search_column=search_column,
|
||||||
)
|
)
|
||||||
filters.extend(search_filters)
|
filters.extend(search_filters)
|
||||||
search_joins.extend(new_search_joins)
|
search_joins.extend(new_search_joins)
|
||||||
@@ -1153,6 +1303,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
filter_attributes = await cls._build_filter_attributes(
|
filter_attributes = await cls._build_filter_attributes(
|
||||||
session, facet_fields, filters, search_joins
|
session, facet_fields, filters, search_joins
|
||||||
)
|
)
|
||||||
|
search_columns = cls._resolve_search_columns(search_fields)
|
||||||
|
|
||||||
return OffsetPaginatedResponse(
|
return OffsetPaginatedResponse(
|
||||||
data=items,
|
data=items,
|
||||||
@@ -1163,6 +1314,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
has_more=has_more,
|
has_more=has_more,
|
||||||
),
|
),
|
||||||
filter_attributes=filter_attributes,
|
filter_attributes=filter_attributes,
|
||||||
|
search_columns=search_columns,
|
||||||
)
|
)
|
||||||
|
|
||||||
@classmethod
|
@classmethod
|
||||||
@@ -1179,6 +1331,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
items_per_page: int = 20,
|
items_per_page: int = 20,
|
||||||
search: str | SearchConfig | None = None,
|
search: str | SearchConfig | None = None,
|
||||||
search_fields: Sequence[SearchFieldType] | None = None,
|
search_fields: Sequence[SearchFieldType] | None = None,
|
||||||
|
search_column: str | None = None,
|
||||||
facet_fields: Sequence[FacetFieldType] | None = None,
|
facet_fields: Sequence[FacetFieldType] | None = None,
|
||||||
filter_by: dict[str, Any] | BaseModel | None = None,
|
filter_by: dict[str, Any] | BaseModel | None = None,
|
||||||
schema: type[BaseModel],
|
schema: type[BaseModel],
|
||||||
@@ -1199,6 +1352,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
items_per_page: Number of items per page (default 20).
|
items_per_page: Number of items per page (default 20).
|
||||||
search: Search query string or SearchConfig object.
|
search: Search query string or SearchConfig object.
|
||||||
search_fields: Fields to search in (overrides class default).
|
search_fields: Fields to search in (overrides class default).
|
||||||
|
search_column: Restrict search to a single column key.
|
||||||
facet_fields: Columns to compute distinct values for (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.
|
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,
|
Keys must match the column.key of a facet field. Scalar → equality,
|
||||||
@@ -1238,6 +1392,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
search,
|
search,
|
||||||
search_fields=search_fields,
|
search_fields=search_fields,
|
||||||
default_fields=cls.searchable_fields,
|
default_fields=cls.searchable_fields,
|
||||||
|
search_column=search_column,
|
||||||
)
|
)
|
||||||
filters.extend(search_filters)
|
filters.extend(search_filters)
|
||||||
search_joins.extend(new_search_joins)
|
search_joins.extend(new_search_joins)
|
||||||
@@ -1308,6 +1463,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
filter_attributes = await cls._build_filter_attributes(
|
filter_attributes = await cls._build_filter_attributes(
|
||||||
session, facet_fields, filters, search_joins
|
session, facet_fields, filters, search_joins
|
||||||
)
|
)
|
||||||
|
search_columns = cls._resolve_search_columns(search_fields)
|
||||||
|
|
||||||
return CursorPaginatedResponse(
|
return CursorPaginatedResponse(
|
||||||
data=items,
|
data=items,
|
||||||
@@ -1318,6 +1474,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
has_more=has_more,
|
has_more=has_more,
|
||||||
),
|
),
|
||||||
filter_attributes=filter_attributes,
|
filter_attributes=filter_attributes,
|
||||||
|
search_columns=search_columns,
|
||||||
)
|
)
|
||||||
|
|
||||||
@overload
|
@overload
|
||||||
@@ -1338,6 +1495,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
include_total: bool = ...,
|
include_total: bool = ...,
|
||||||
search: str | SearchConfig | None = ...,
|
search: str | SearchConfig | None = ...,
|
||||||
search_fields: Sequence[SearchFieldType] | None = ...,
|
search_fields: Sequence[SearchFieldType] | None = ...,
|
||||||
|
search_column: str | None = ...,
|
||||||
facet_fields: Sequence[FacetFieldType] | None = ...,
|
facet_fields: Sequence[FacetFieldType] | None = ...,
|
||||||
filter_by: dict[str, Any] | BaseModel | None = ...,
|
filter_by: dict[str, Any] | BaseModel | None = ...,
|
||||||
schema: type[BaseModel],
|
schema: type[BaseModel],
|
||||||
@@ -1361,6 +1519,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
include_total: bool = ...,
|
include_total: bool = ...,
|
||||||
search: str | SearchConfig | None = ...,
|
search: str | SearchConfig | None = ...,
|
||||||
search_fields: Sequence[SearchFieldType] | None = ...,
|
search_fields: Sequence[SearchFieldType] | None = ...,
|
||||||
|
search_column: str | None = ...,
|
||||||
facet_fields: Sequence[FacetFieldType] | None = ...,
|
facet_fields: Sequence[FacetFieldType] | None = ...,
|
||||||
filter_by: dict[str, Any] | BaseModel | None = ...,
|
filter_by: dict[str, Any] | BaseModel | None = ...,
|
||||||
schema: type[BaseModel],
|
schema: type[BaseModel],
|
||||||
@@ -1383,6 +1542,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
include_total: bool = True,
|
include_total: bool = True,
|
||||||
search: str | SearchConfig | None = None,
|
search: str | SearchConfig | None = None,
|
||||||
search_fields: Sequence[SearchFieldType] | None = None,
|
search_fields: Sequence[SearchFieldType] | None = None,
|
||||||
|
search_column: str | None = None,
|
||||||
facet_fields: Sequence[FacetFieldType] | None = None,
|
facet_fields: Sequence[FacetFieldType] | None = None,
|
||||||
filter_by: dict[str, Any] | BaseModel | None = None,
|
filter_by: dict[str, Any] | BaseModel | None = None,
|
||||||
schema: type[BaseModel],
|
schema: type[BaseModel],
|
||||||
@@ -1410,6 +1570,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
only applies when ``pagination_type`` is ``OFFSET``.
|
only applies when ``pagination_type`` is ``OFFSET``.
|
||||||
search: Search query string or :class:`.SearchConfig` object.
|
search: Search query string or :class:`.SearchConfig` object.
|
||||||
search_fields: Fields to search in (overrides class default).
|
search_fields: Fields to search in (overrides class default).
|
||||||
|
search_column: Restrict search to a single column key.
|
||||||
facet_fields: Columns to compute distinct values for (overrides
|
facet_fields: Columns to compute distinct values for (overrides
|
||||||
class default).
|
class default).
|
||||||
filter_by: Dict of ``{column_key: value}`` to filter by declared
|
filter_by: Dict of ``{column_key: value}`` to filter by declared
|
||||||
@@ -1438,6 +1599,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
items_per_page=items_per_page,
|
items_per_page=items_per_page,
|
||||||
search=search,
|
search=search,
|
||||||
search_fields=search_fields,
|
search_fields=search_fields,
|
||||||
|
search_column=search_column,
|
||||||
facet_fields=facet_fields,
|
facet_fields=facet_fields,
|
||||||
filter_by=filter_by,
|
filter_by=filter_by,
|
||||||
schema=schema,
|
schema=schema,
|
||||||
@@ -1457,6 +1619,7 @@ class AsyncCrud(Generic[ModelType]):
|
|||||||
include_total=include_total,
|
include_total=include_total,
|
||||||
search=search,
|
search=search,
|
||||||
search_fields=search_fields,
|
search_fields=search_fields,
|
||||||
|
search_column=search_column,
|
||||||
facet_fields=facet_fields,
|
facet_fields=facet_fields,
|
||||||
filter_by=filter_by,
|
filter_by=filter_by,
|
||||||
schema=schema,
|
schema=schema,
|
||||||
@@ -1488,7 +1651,7 @@ def CrudFactory(
|
|||||||
responses. Supports direct columns (``User.status``) and relationship tuples
|
responses. Supports direct columns (``User.status``) and relationship tuples
|
||||||
(``(User.role, Role.name)``). Can be overridden per call.
|
(``(User.role, Role.name)``). Can be overridden per call.
|
||||||
order_fields: Optional list of model attributes that callers are allowed to order by
|
order_fields: Optional list of model attributes that callers are allowed to order by
|
||||||
via ``order_params()``. Can be overridden per call.
|
via ``offset_paginate_params()``. Can be overridden per call.
|
||||||
m2m_fields: Optional mapping for many-to-many relationships.
|
m2m_fields: Optional mapping for many-to-many relationships.
|
||||||
Maps schema field names (containing lists of IDs) to
|
Maps schema field names (containing lists of IDs) to
|
||||||
SQLAlchemy relationship attributes.
|
SQLAlchemy relationship attributes.
|
||||||
|
|||||||
@@ -24,6 +24,7 @@ from sqlalchemy.types import (
|
|||||||
|
|
||||||
from ..exceptions import (
|
from ..exceptions import (
|
||||||
InvalidFacetFilterError,
|
InvalidFacetFilterError,
|
||||||
|
InvalidSearchColumnError,
|
||||||
NoSearchableFieldsError,
|
NoSearchableFieldsError,
|
||||||
UnsupportedFacetTypeError,
|
UnsupportedFacetTypeError,
|
||||||
)
|
)
|
||||||
@@ -96,6 +97,7 @@ def build_search_filters(
|
|||||||
search: str | SearchConfig,
|
search: str | SearchConfig,
|
||||||
search_fields: Sequence[SearchFieldType] | None = None,
|
search_fields: Sequence[SearchFieldType] | None = None,
|
||||||
default_fields: Sequence[SearchFieldType] | None = None,
|
default_fields: Sequence[SearchFieldType] | None = None,
|
||||||
|
search_column: str | None = None,
|
||||||
) -> tuple[list["ColumnElement[bool]"], list[InstrumentedAttribute[Any]]]:
|
) -> tuple[list["ColumnElement[bool]"], list[InstrumentedAttribute[Any]]]:
|
||||||
"""Build SQLAlchemy filter conditions for search.
|
"""Build SQLAlchemy filter conditions for search.
|
||||||
|
|
||||||
@@ -104,6 +106,8 @@ def build_search_filters(
|
|||||||
search: Search string or SearchConfig
|
search: Search string or SearchConfig
|
||||||
search_fields: Fields specified per-call (takes priority)
|
search_fields: Fields specified per-call (takes priority)
|
||||||
default_fields: Default fields (from ClassVar)
|
default_fields: Default fields (from ClassVar)
|
||||||
|
search_column: Optional key to narrow search to a single field.
|
||||||
|
Must match one of the resolved search field keys.
|
||||||
|
|
||||||
Returns:
|
Returns:
|
||||||
Tuple of (filter_conditions, joins_needed)
|
Tuple of (filter_conditions, joins_needed)
|
||||||
@@ -130,6 +134,14 @@ def build_search_filters(
|
|||||||
if not fields:
|
if not fields:
|
||||||
raise NoSearchableFieldsError(model)
|
raise NoSearchableFieldsError(model)
|
||||||
|
|
||||||
|
# Narrow to a single column when search_column is specified
|
||||||
|
if search_column is not None:
|
||||||
|
keys = search_field_keys(fields)
|
||||||
|
index = {k: f for k, f in zip(keys, fields)}
|
||||||
|
if search_column not in index:
|
||||||
|
raise InvalidSearchColumnError(search_column, sorted(index))
|
||||||
|
fields = [index[search_column]]
|
||||||
|
|
||||||
query = config.query.strip()
|
query = config.query.strip()
|
||||||
filters: list[ColumnElement[bool]] = []
|
filters: list[ColumnElement[bool]] = []
|
||||||
joins: list[InstrumentedAttribute[Any]] = []
|
joins: list[InstrumentedAttribute[Any]] = []
|
||||||
@@ -164,6 +176,11 @@ def build_search_filters(
|
|||||||
return filters, joins
|
return filters, joins
|
||||||
|
|
||||||
|
|
||||||
|
def search_field_keys(fields: Sequence[SearchFieldType]) -> list[str]:
|
||||||
|
"""Return a human-readable key for each search field."""
|
||||||
|
return facet_keys(fields)
|
||||||
|
|
||||||
|
|
||||||
def facet_keys(facet_fields: Sequence[FacetFieldType]) -> list[str]:
|
def facet_keys(facet_fields: Sequence[FacetFieldType]) -> list[str]:
|
||||||
"""Return a key for each facet field.
|
"""Return a key for each facet field.
|
||||||
|
|
||||||
|
|||||||
@@ -7,6 +7,7 @@ from .exceptions import (
|
|||||||
ForbiddenError,
|
ForbiddenError,
|
||||||
InvalidFacetFilterError,
|
InvalidFacetFilterError,
|
||||||
InvalidOrderFieldError,
|
InvalidOrderFieldError,
|
||||||
|
InvalidSearchColumnError,
|
||||||
NoSearchableFieldsError,
|
NoSearchableFieldsError,
|
||||||
NotFoundError,
|
NotFoundError,
|
||||||
UnauthorizedError,
|
UnauthorizedError,
|
||||||
@@ -24,6 +25,7 @@ __all__ = [
|
|||||||
"init_exceptions_handlers",
|
"init_exceptions_handlers",
|
||||||
"InvalidFacetFilterError",
|
"InvalidFacetFilterError",
|
||||||
"InvalidOrderFieldError",
|
"InvalidOrderFieldError",
|
||||||
|
"InvalidSearchColumnError",
|
||||||
"NoSearchableFieldsError",
|
"NoSearchableFieldsError",
|
||||||
"NotFoundError",
|
"NotFoundError",
|
||||||
"UnauthorizedError",
|
"UnauthorizedError",
|
||||||
|
|||||||
@@ -172,6 +172,33 @@ class UnsupportedFacetTypeError(ApiException):
|
|||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
class InvalidSearchColumnError(ApiException):
|
||||||
|
"""Raised when search_column is not one of the configured searchable fields."""
|
||||||
|
|
||||||
|
api_error = ApiError(
|
||||||
|
code=400,
|
||||||
|
msg="Invalid Search Column",
|
||||||
|
desc="The requested search column is not a configured searchable field.",
|
||||||
|
err_code="SEARCH-COL-400",
|
||||||
|
)
|
||||||
|
|
||||||
|
def __init__(self, column: str, valid_columns: list[str]) -> None:
|
||||||
|
"""Initialize the exception.
|
||||||
|
|
||||||
|
Args:
|
||||||
|
column: The unknown search column provided by the caller.
|
||||||
|
valid_columns: List of valid search column keys.
|
||||||
|
"""
|
||||||
|
self.column = column
|
||||||
|
self.valid_columns = valid_columns
|
||||||
|
super().__init__(
|
||||||
|
desc=(
|
||||||
|
f"'{column}' is not a searchable column. "
|
||||||
|
f"Valid columns: {valid_columns}."
|
||||||
|
)
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
class InvalidOrderFieldError(ApiException):
|
class InvalidOrderFieldError(ApiException):
|
||||||
"""Raised when order_by contains a field not in the allowed order fields."""
|
"""Raised when order_by contains a field not in the allowed order fields."""
|
||||||
|
|
||||||
|
|||||||
@@ -122,7 +122,7 @@ def _format_validation_error(
|
|||||||
)
|
)
|
||||||
|
|
||||||
return JSONResponse(
|
return JSONResponse(
|
||||||
status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
|
status_code=status.HTTP_422_UNPROCESSABLE_CONTENT,
|
||||||
content=error_response.model_dump(),
|
content=error_response.model_dump(),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|||||||
@@ -1,6 +1,6 @@
|
|||||||
"""Prometheus metrics endpoint for FastAPI applications."""
|
"""Prometheus metrics endpoint for FastAPI applications."""
|
||||||
|
|
||||||
import asyncio
|
import inspect
|
||||||
import os
|
import os
|
||||||
|
|
||||||
from fastapi import FastAPI
|
from fastapi import FastAPI
|
||||||
@@ -55,10 +55,10 @@ def init_metrics(
|
|||||||
|
|
||||||
# Partition collectors and cache env check at startup — both are stable for the app lifetime.
|
# Partition collectors and cache env check at startup — both are stable for the app lifetime.
|
||||||
async_collectors = [
|
async_collectors = [
|
||||||
c for c in registry.get_collectors() if asyncio.iscoroutinefunction(c.func)
|
c for c in registry.get_collectors() if inspect.iscoroutinefunction(c.func)
|
||||||
]
|
]
|
||||||
sync_collectors = [
|
sync_collectors = [
|
||||||
c for c in registry.get_collectors() if not asyncio.iscoroutinefunction(c.func)
|
c for c in registry.get_collectors() if not inspect.iscoroutinefunction(c.func)
|
||||||
]
|
]
|
||||||
multiprocess_mode = _is_multiprocess()
|
multiprocess_mode = _is_multiprocess()
|
||||||
|
|
||||||
|
|||||||
@@ -1,7 +1,6 @@
|
|||||||
"""Pytest helper utilities for FastAPI testing."""
|
"""Pytest helper utilities for FastAPI testing."""
|
||||||
|
|
||||||
import os
|
import os
|
||||||
import warnings
|
|
||||||
from collections.abc import AsyncGenerator, Callable
|
from collections.abc import AsyncGenerator, Callable
|
||||||
from contextlib import asynccontextmanager
|
from contextlib import asynccontextmanager
|
||||||
from typing import Any
|
from typing import Any
|
||||||
@@ -16,31 +15,10 @@ from sqlalchemy.ext.asyncio import (
|
|||||||
)
|
)
|
||||||
from sqlalchemy.orm import DeclarativeBase
|
from sqlalchemy.orm import DeclarativeBase
|
||||||
|
|
||||||
from ..db import cleanup_tables as _cleanup_tables
|
from ..db import cleanup_tables, create_database
|
||||||
from ..db import create_database
|
|
||||||
from ..models.watched import EventSession
|
from ..models.watched import EventSession
|
||||||
|
|
||||||
|
|
||||||
async def cleanup_tables(
|
|
||||||
session: AsyncSession,
|
|
||||||
base: type[DeclarativeBase],
|
|
||||||
) -> None:
|
|
||||||
"""Truncate all tables for fast between-test cleanup.
|
|
||||||
|
|
||||||
.. deprecated::
|
|
||||||
Import ``cleanup_tables`` from ``fastapi_toolsets.db`` instead.
|
|
||||||
This re-export will be removed in v3.0.0.
|
|
||||||
"""
|
|
||||||
warnings.warn(
|
|
||||||
"Importing cleanup_tables from fastapi_toolsets.pytest is deprecated "
|
|
||||||
"and will be removed in v3.0.0. "
|
|
||||||
"Use 'from fastapi_toolsets.db import cleanup_tables' instead.",
|
|
||||||
DeprecationWarning,
|
|
||||||
stacklevel=2,
|
|
||||||
)
|
|
||||||
await _cleanup_tables(session=session, base=base)
|
|
||||||
|
|
||||||
|
|
||||||
def _get_xdist_worker(default_test_db: str) -> str:
|
def _get_xdist_worker(default_test_db: str) -> str:
|
||||||
"""Return the pytest-xdist worker name, or *default_test_db* when not running under xdist.
|
"""Return the pytest-xdist worker name, or *default_test_db* when not running under xdist.
|
||||||
|
|
||||||
@@ -273,7 +251,7 @@ async def create_db_session(
|
|||||||
yield session
|
yield session
|
||||||
|
|
||||||
if cleanup:
|
if cleanup:
|
||||||
await _cleanup_tables(session=session, base=base)
|
await cleanup_tables(session=session, base=base)
|
||||||
|
|
||||||
if drop_tables:
|
if drop_tables:
|
||||||
async with engine.begin() as conn:
|
async with engine.begin() as conn:
|
||||||
|
|||||||
@@ -162,6 +162,7 @@ class PaginatedResponse(BaseResponse, Generic[DataT]):
|
|||||||
pagination: OffsetPagination | CursorPagination
|
pagination: OffsetPagination | CursorPagination
|
||||||
pagination_type: PaginationType | None = None
|
pagination_type: PaginationType | None = None
|
||||||
filter_attributes: dict[str, list[Any]] | None = None
|
filter_attributes: dict[str, list[Any]] | None = None
|
||||||
|
search_columns: list[str] | None = None
|
||||||
|
|
||||||
_discriminated_union_cache: ClassVar[dict[Any, Any]] = {}
|
_discriminated_union_cache: ClassVar[dict[Any, Any]] = {}
|
||||||
|
|
||||||
|
|||||||
@@ -211,6 +211,38 @@ class TestResolveLoadOptions:
|
|||||||
assert crud._resolve_load_options([]) == []
|
assert crud._resolve_load_options([]) == []
|
||||||
|
|
||||||
|
|
||||||
|
class TestResolveSearchColumns:
|
||||||
|
"""Tests for _resolve_search_columns logic."""
|
||||||
|
|
||||||
|
def test_returns_none_when_no_searchable_fields(self):
|
||||||
|
"""Returns None when cls.searchable_fields is None and no search_fields passed."""
|
||||||
|
|
||||||
|
class AbstractCrud(AsyncCrud[User]):
|
||||||
|
pass
|
||||||
|
|
||||||
|
assert AbstractCrud._resolve_search_columns(None) is None
|
||||||
|
|
||||||
|
def test_returns_none_when_empty_search_fields_passed(self):
|
||||||
|
"""Returns None when an empty list is passed explicitly."""
|
||||||
|
crud = CrudFactory(User)
|
||||||
|
assert crud._resolve_search_columns([]) is None
|
||||||
|
|
||||||
|
def test_returns_keys_from_class_searchable_fields(self):
|
||||||
|
"""Returns column keys from cls.searchable_fields when no override passed."""
|
||||||
|
crud = CrudFactory(User, searchable_fields=[User.username])
|
||||||
|
result = crud._resolve_search_columns(None)
|
||||||
|
assert result is not None
|
||||||
|
assert "username" in result
|
||||||
|
|
||||||
|
def test_search_fields_override_takes_priority(self):
|
||||||
|
"""Explicit search_fields override cls.searchable_fields."""
|
||||||
|
crud = CrudFactory(User, searchable_fields=[User.username])
|
||||||
|
result = crud._resolve_search_columns([User.email])
|
||||||
|
assert result is not None
|
||||||
|
assert "email" in result
|
||||||
|
assert "username" not in result
|
||||||
|
|
||||||
|
|
||||||
class TestDefaultLoadOptionsIntegration:
|
class TestDefaultLoadOptionsIntegration:
|
||||||
"""Integration tests for default_load_options with real DB queries."""
|
"""Integration tests for default_load_options with real DB queries."""
|
||||||
|
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ from sqlalchemy.sql.elements import ColumnElement, UnaryExpression
|
|||||||
from fastapi_toolsets.crud import (
|
from fastapi_toolsets.crud import (
|
||||||
CrudFactory,
|
CrudFactory,
|
||||||
InvalidFacetFilterError,
|
InvalidFacetFilterError,
|
||||||
|
InvalidSearchColumnError,
|
||||||
SearchConfig,
|
SearchConfig,
|
||||||
UnsupportedFacetTypeError,
|
UnsupportedFacetTypeError,
|
||||||
get_searchable_fields,
|
get_searchable_fields,
|
||||||
@@ -1030,61 +1031,64 @@ class TestFilterBy:
|
|||||||
assert "JSON" in exc_info.value.col_type
|
assert "JSON" in exc_info.value.col_type
|
||||||
|
|
||||||
|
|
||||||
class TestFilterParamsSchema:
|
class TestFilterParamsViaConsolidated:
|
||||||
"""Tests for AsyncCrud.filter_params()."""
|
"""Tests for filter params via consolidated offset_paginate_params()."""
|
||||||
|
|
||||||
def test_generates_fields_from_facet_fields(self):
|
def test_generates_fields_from_facet_fields(self):
|
||||||
"""Returned dependency has one keyword param per facet field."""
|
"""Returned dependency has one keyword param per facet field."""
|
||||||
import inspect
|
|
||||||
|
|
||||||
UserFacetCrud = CrudFactory(User, facet_fields=[User.username, User.email])
|
UserFacetCrud = CrudFactory(User, facet_fields=[User.username, User.email])
|
||||||
dep = UserFacetCrud.filter_params()
|
dep = UserFacetCrud.offset_paginate_params(search=False, order=False)
|
||||||
|
|
||||||
param_names = set(inspect.signature(dep).parameters)
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
assert param_names == {"username", "email"}
|
assert "username" in param_names
|
||||||
|
assert "email" in param_names
|
||||||
|
|
||||||
def test_relationship_facet_uses_full_chain_key(self):
|
def test_relationship_facet_uses_full_chain_key(self):
|
||||||
"""Relationship tuple uses the full chain joined by __ as the key."""
|
"""Relationship tuple uses the full chain joined by __ as the key."""
|
||||||
import inspect
|
|
||||||
|
|
||||||
UserRoleCrud = CrudFactory(User, facet_fields=[(User.role, Role.name)])
|
UserRoleCrud = CrudFactory(User, facet_fields=[(User.role, Role.name)])
|
||||||
dep = UserRoleCrud.filter_params()
|
dep = UserRoleCrud.offset_paginate_params(search=False, order=False)
|
||||||
|
|
||||||
param_names = set(inspect.signature(dep).parameters)
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
assert param_names == {"role__name"}
|
assert "role__name" in param_names
|
||||||
|
|
||||||
def test_raises_when_no_facet_fields(self):
|
def test_filter_disabled_no_facet_params(self):
|
||||||
"""ValueError raised when no facet_fields are configured or provided."""
|
"""When filter=False, no facet params are generated."""
|
||||||
with pytest.raises(ValueError, match="no facet_fields"):
|
UserFacetCrud = CrudFactory(User, facet_fields=[User.username, User.email])
|
||||||
UserCrud.filter_params()
|
dep = UserFacetCrud.offset_paginate_params(
|
||||||
|
search=False, filter=False, order=False
|
||||||
|
)
|
||||||
|
|
||||||
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
|
assert param_names == {"page", "items_per_page"}
|
||||||
|
|
||||||
def test_facet_fields_override(self):
|
def test_facet_fields_override(self):
|
||||||
"""facet_fields= parameter overrides the class-level default."""
|
"""facet_fields= parameter overrides the class-level default."""
|
||||||
import inspect
|
|
||||||
|
|
||||||
UserFacetCrud = CrudFactory(User, facet_fields=[User.username, User.email])
|
UserFacetCrud = CrudFactory(User, facet_fields=[User.username, User.email])
|
||||||
dep = UserFacetCrud.filter_params(facet_fields=[User.email])
|
dep = UserFacetCrud.offset_paginate_params(
|
||||||
|
search=False, order=False, facet_fields=[User.email]
|
||||||
|
)
|
||||||
|
|
||||||
param_names = set(inspect.signature(dep).parameters)
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
assert param_names == {"email"}
|
assert "email" in param_names
|
||||||
|
assert "username" not in param_names
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_awaiting_dep_returns_dict_with_values(self):
|
async def test_awaiting_dep_returns_filter_by_with_values(self):
|
||||||
"""Awaiting the dependency returns a dict with only the supplied keys."""
|
"""Awaiting the dependency returns filter_by dict with only supplied keys."""
|
||||||
UserFacetCrud = CrudFactory(User, facet_fields=[User.username, User.email])
|
UserFacetCrud = CrudFactory(User, facet_fields=[User.username, User.email])
|
||||||
dep = UserFacetCrud.filter_params()
|
dep = UserFacetCrud.offset_paginate_params(search=False, order=False)
|
||||||
|
|
||||||
result = await dep(username=["alice"])
|
result = await dep(page=1, items_per_page=20, username=["alice"], email=None)
|
||||||
assert result == {"username": ["alice"]}
|
assert result["filter_by"] == {"username": ["alice"]}
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_multi_value_list_field(self):
|
async def test_multi_value_list_field(self):
|
||||||
"""Multiple values are accepted as a list."""
|
"""Multiple values are accepted as a list."""
|
||||||
UserFacetCrud = CrudFactory(User, facet_fields=[User.username])
|
UserFacetCrud = CrudFactory(User, facet_fields=[User.username])
|
||||||
dep = UserFacetCrud.filter_params()
|
dep = UserFacetCrud.offset_paginate_params(search=False, order=False)
|
||||||
|
|
||||||
result = await dep(username=["alice", "bob"])
|
result = await dep(page=1, items_per_page=20, username=["alice", "bob"])
|
||||||
assert result == {"username": ["alice", "bob"]}
|
assert result["filter_by"] == {"username": ["alice", "bob"]}
|
||||||
|
|
||||||
def test_disambiguates_duplicate_column_keys(self):
|
def test_disambiguates_duplicate_column_keys(self):
|
||||||
"""Two relationship tuples sharing a terminal column key get prefixed names."""
|
"""Two relationship tuples sharing a terminal column key get prefixed names."""
|
||||||
@@ -1129,15 +1133,15 @@ class TestFilterParamsSchema:
|
|||||||
assert keys == ["username", "email"]
|
assert keys == ["username", "email"]
|
||||||
|
|
||||||
def test_dependency_name_includes_model_name(self):
|
def test_dependency_name_includes_model_name(self):
|
||||||
"""Returned dependency is named {Model}FilterParams."""
|
"""Returned dependency is named {Model}OffsetPaginateParams."""
|
||||||
UserFacetCrud = CrudFactory(User, facet_fields=[User.username])
|
UserFacetCrud = CrudFactory(User, facet_fields=[User.username])
|
||||||
dep = UserFacetCrud.filter_params()
|
dep = UserFacetCrud.offset_paginate_params(search=False, order=False)
|
||||||
|
|
||||||
assert dep.__name__ == "UserFilterParams" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
|
assert dep.__name__ == "UserOffsetPaginateParams" # type: ignore[union-attr] # ty:ignore[unresolved-attribute]
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_integration_with_offset_paginate(self, db_session: AsyncSession):
|
async def test_integration_with_offset_paginate(self, db_session: AsyncSession):
|
||||||
"""Dependency result can be passed directly to offset_paginate via filter_by."""
|
"""Dependency result can be unpacked directly into offset_paginate."""
|
||||||
UserFacetCrud = CrudFactory(User, facet_fields=[User.username])
|
UserFacetCrud = CrudFactory(User, facet_fields=[User.username])
|
||||||
await UserCrud.create(
|
await UserCrud.create(
|
||||||
db_session, UserCreate(username="alice", email="a@test.com")
|
db_session, UserCreate(username="alice", email="a@test.com")
|
||||||
@@ -1146,10 +1150,10 @@ class TestFilterParamsSchema:
|
|||||||
db_session, UserCreate(username="bob", email="b@test.com")
|
db_session, UserCreate(username="bob", email="b@test.com")
|
||||||
)
|
)
|
||||||
|
|
||||||
dep = UserFacetCrud.filter_params()
|
dep = UserFacetCrud.offset_paginate_params(search=False, order=False)
|
||||||
f = await dep(username=["alice"])
|
params = await dep(page=1, items_per_page=20, username=["alice"])
|
||||||
result = await UserFacetCrud.offset_paginate(
|
result = await UserFacetCrud.offset_paginate(
|
||||||
db_session, filter_by=f, schema=UserRead
|
db_session, **params, schema=UserRead
|
||||||
)
|
)
|
||||||
|
|
||||||
assert isinstance(result.pagination, OffsetPagination)
|
assert isinstance(result.pagination, OffsetPagination)
|
||||||
@@ -1158,7 +1162,7 @@ class TestFilterParamsSchema:
|
|||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_dep_result_passed_to_cursor_paginate(self, db_session: AsyncSession):
|
async def test_dep_result_passed_to_cursor_paginate(self, db_session: AsyncSession):
|
||||||
"""Dependency result can be passed directly to cursor_paginate via filter_by."""
|
"""Dependency result can be unpacked directly into cursor_paginate."""
|
||||||
UserFacetCursorCrud = CrudFactory(
|
UserFacetCursorCrud = CrudFactory(
|
||||||
User, cursor_column=User.id, facet_fields=[User.username]
|
User, cursor_column=User.id, facet_fields=[User.username]
|
||||||
)
|
)
|
||||||
@@ -1169,10 +1173,10 @@ class TestFilterParamsSchema:
|
|||||||
db_session, UserCreate(username="bob", email="b@test.com")
|
db_session, UserCreate(username="bob", email="b@test.com")
|
||||||
)
|
)
|
||||||
|
|
||||||
dep = UserFacetCursorCrud.filter_params()
|
dep = UserFacetCursorCrud.cursor_paginate_params(search=False, order=False)
|
||||||
f = await dep(username=["alice"])
|
params = await dep(cursor=None, items_per_page=20, username=["alice"])
|
||||||
result = await UserFacetCursorCrud.cursor_paginate(
|
result = await UserFacetCursorCrud.cursor_paginate(
|
||||||
db_session, filter_by=f, schema=UserRead
|
db_session, **params, schema=UserRead
|
||||||
)
|
)
|
||||||
|
|
||||||
assert len(result.data) == 1
|
assert len(result.data) == 1
|
||||||
@@ -1180,7 +1184,7 @@ class TestFilterParamsSchema:
|
|||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_all_none_dep_result_passes_no_filter(self, db_session: AsyncSession):
|
async def test_all_none_dep_result_passes_no_filter(self, db_session: AsyncSession):
|
||||||
"""All-None dependency result results in no filter (returns all rows)."""
|
"""All-None dependency result results in filter_by=None (returns all rows)."""
|
||||||
UserFacetCrud = CrudFactory(User, facet_fields=[User.username])
|
UserFacetCrud = CrudFactory(User, facet_fields=[User.username])
|
||||||
await UserCrud.create(
|
await UserCrud.create(
|
||||||
db_session, UserCreate(username="alice", email="a@test.com")
|
db_session, UserCreate(username="alice", email="a@test.com")
|
||||||
@@ -1189,46 +1193,269 @@ class TestFilterParamsSchema:
|
|||||||
db_session, UserCreate(username="bob", email="b@test.com")
|
db_session, UserCreate(username="bob", email="b@test.com")
|
||||||
)
|
)
|
||||||
|
|
||||||
dep = UserFacetCrud.filter_params()
|
dep = UserFacetCrud.offset_paginate_params(search=False, order=False)
|
||||||
f = await dep() # all fields None
|
params = await dep(page=1, items_per_page=20) # all facet fields None
|
||||||
|
assert params["filter_by"] is None
|
||||||
result = await UserFacetCrud.offset_paginate(
|
result = await UserFacetCrud.offset_paginate(
|
||||||
db_session, filter_by=f, schema=UserRead
|
db_session, **params, schema=UserRead
|
||||||
)
|
)
|
||||||
|
|
||||||
assert isinstance(result.pagination, OffsetPagination)
|
assert isinstance(result.pagination, OffsetPagination)
|
||||||
assert result.pagination.total_count == 2
|
assert result.pagination.total_count == 2
|
||||||
|
|
||||||
|
def test_facet_key_collision_raises(self):
|
||||||
|
"""ValueError raised when a facet key clashes with a reserved param name."""
|
||||||
|
# Create a mock facet field whose key would be "search"
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
class TestOrderParamsSchema:
|
mock_field = MagicMock()
|
||||||
"""Tests for AsyncCrud.order_params()."""
|
mock_field.key = "search"
|
||||||
|
mock_field.property.columns = [MagicMock()]
|
||||||
|
|
||||||
|
UserFacetCrud = CrudFactory(User, facet_fields=[mock_field])
|
||||||
|
with pytest.raises(ValueError, match="conflicts with a reserved"):
|
||||||
|
UserFacetCrud.offset_paginate_params(search=True, order=False)
|
||||||
|
|
||||||
|
|
||||||
|
class TestSearchParamsViaConsolidated:
|
||||||
|
"""Tests for search params via consolidated offset_paginate_params()."""
|
||||||
|
|
||||||
|
def test_generates_search_and_search_column_params(self):
|
||||||
|
"""Returned dependency has search and search_column query params."""
|
||||||
|
UserSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, User.email]
|
||||||
|
)
|
||||||
|
dep = UserSearchCrud.offset_paginate_params(filter=False, order=False)
|
||||||
|
|
||||||
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
|
assert "search" in param_names
|
||||||
|
assert "search_column" in param_names
|
||||||
|
|
||||||
|
def test_search_disabled_no_search_params(self):
|
||||||
|
"""When search=False, no search params are generated."""
|
||||||
|
UserSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, User.email]
|
||||||
|
)
|
||||||
|
dep = UserSearchCrud.offset_paginate_params(
|
||||||
|
search=False, filter=False, order=False
|
||||||
|
)
|
||||||
|
|
||||||
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
|
assert "search" not in param_names
|
||||||
|
assert "search_column" not in param_names
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_awaiting_dep_with_search_only(self):
|
||||||
|
"""Awaiting the dependency with only search returns search in dict."""
|
||||||
|
UserSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, User.email]
|
||||||
|
)
|
||||||
|
dep = UserSearchCrud.offset_paginate_params(filter=False, order=False)
|
||||||
|
|
||||||
|
result = await dep(
|
||||||
|
page=1, items_per_page=20, search="alice", search_column=None
|
||||||
|
)
|
||||||
|
assert result["search"] == "alice"
|
||||||
|
assert "search_column" not in result
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_awaiting_dep_with_search_and_column(self):
|
||||||
|
"""Awaiting the dependency with both params returns both keys."""
|
||||||
|
UserSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, User.email]
|
||||||
|
)
|
||||||
|
dep = UserSearchCrud.offset_paginate_params(filter=False, order=False)
|
||||||
|
|
||||||
|
result = await dep(
|
||||||
|
page=1, items_per_page=20, search="alice", search_column="username"
|
||||||
|
)
|
||||||
|
assert result["search"] == "alice"
|
||||||
|
assert result["search_column"] == "username"
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_awaiting_dep_with_no_search_values(self):
|
||||||
|
"""Awaiting the dependency with no search values omits search keys."""
|
||||||
|
UserSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, User.email]
|
||||||
|
)
|
||||||
|
dep = UserSearchCrud.offset_paginate_params(filter=False, order=False)
|
||||||
|
|
||||||
|
result = await dep(page=1, items_per_page=20, search=None, search_column=None)
|
||||||
|
assert "search" not in result
|
||||||
|
assert "search_column" not in result
|
||||||
|
|
||||||
|
def test_relationship_search_field_key(self):
|
||||||
|
"""Relationship tuple search fields use __ joined keys."""
|
||||||
|
UserRelSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, (User.role, Role.name)]
|
||||||
|
)
|
||||||
|
dep = UserRelSearchCrud.offset_paginate_params(filter=False, order=False)
|
||||||
|
|
||||||
|
params = inspect.signature(dep).parameters
|
||||||
|
search_column_param = params["search_column"]
|
||||||
|
assert search_column_param.default.json_schema_extra.get("enum") == [
|
||||||
|
"id",
|
||||||
|
"username",
|
||||||
|
"role__name",
|
||||||
|
]
|
||||||
|
|
||||||
|
|
||||||
|
class TestSearchColumns:
|
||||||
|
"""Tests for search_columns in paginated responses."""
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_search_columns_returned_in_offset_paginate(
|
||||||
|
self, db_session: AsyncSession
|
||||||
|
):
|
||||||
|
"""offset_paginate response includes search_columns."""
|
||||||
|
UserSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, User.email]
|
||||||
|
)
|
||||||
|
await UserCrud.create(
|
||||||
|
db_session, UserCreate(username="alice", email="a@test.com")
|
||||||
|
)
|
||||||
|
|
||||||
|
result = await UserSearchCrud.offset_paginate(db_session, schema=UserRead)
|
||||||
|
|
||||||
|
assert result.search_columns is not None
|
||||||
|
assert "username" in result.search_columns
|
||||||
|
assert "email" in result.search_columns
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_search_columns_returned_in_cursor_paginate(
|
||||||
|
self, db_session: AsyncSession
|
||||||
|
):
|
||||||
|
"""cursor_paginate response includes search_columns."""
|
||||||
|
UserSearchCursorCrud = CrudFactory(
|
||||||
|
User,
|
||||||
|
cursor_column=User.id,
|
||||||
|
searchable_fields=[User.username, User.email],
|
||||||
|
)
|
||||||
|
await UserCrud.create(
|
||||||
|
db_session, UserCreate(username="alice", email="a@test.com")
|
||||||
|
)
|
||||||
|
|
||||||
|
result = await UserSearchCursorCrud.cursor_paginate(db_session, schema=UserRead)
|
||||||
|
|
||||||
|
assert result.search_columns is not None
|
||||||
|
assert "username" in result.search_columns
|
||||||
|
assert "email" in result.search_columns
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_search_column_narrows_search(self, db_session: AsyncSession):
|
||||||
|
"""search_column restricts search to a single field."""
|
||||||
|
UserSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, User.email]
|
||||||
|
)
|
||||||
|
await UserCrud.create(
|
||||||
|
db_session, UserCreate(username="alice", email="bob@test.com")
|
||||||
|
)
|
||||||
|
await UserCrud.create(
|
||||||
|
db_session, UserCreate(username="bob", email="alice@test.com")
|
||||||
|
)
|
||||||
|
|
||||||
|
# Search "alice" in username only — should return only alice
|
||||||
|
result = await UserSearchCrud.offset_paginate(
|
||||||
|
db_session, search="alice", search_column="username", schema=UserRead
|
||||||
|
)
|
||||||
|
|
||||||
|
assert isinstance(result.pagination, OffsetPagination)
|
||||||
|
assert result.pagination.total_count == 1
|
||||||
|
assert result.data[0].username == "alice"
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_search_column_invalid_raises(self, db_session: AsyncSession):
|
||||||
|
"""search_column with an invalid key raises InvalidSearchColumnError."""
|
||||||
|
UserSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, User.email]
|
||||||
|
)
|
||||||
|
|
||||||
|
with pytest.raises(InvalidSearchColumnError) as exc_info:
|
||||||
|
await UserSearchCrud.offset_paginate(
|
||||||
|
db_session,
|
||||||
|
search="alice",
|
||||||
|
search_column="nonexistent",
|
||||||
|
schema=UserRead,
|
||||||
|
)
|
||||||
|
|
||||||
|
assert exc_info.value.column == "nonexistent"
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_search_without_search_column_searches_all(
|
||||||
|
self, db_session: AsyncSession
|
||||||
|
):
|
||||||
|
"""search without search_column searches across all configured fields."""
|
||||||
|
UserSearchCrud = CrudFactory(
|
||||||
|
User, searchable_fields=[User.username, User.email]
|
||||||
|
)
|
||||||
|
await UserCrud.create(
|
||||||
|
db_session, UserCreate(username="alice", email="bob@test.com")
|
||||||
|
)
|
||||||
|
await UserCrud.create(
|
||||||
|
db_session, UserCreate(username="bob", email="alice@test.com")
|
||||||
|
)
|
||||||
|
|
||||||
|
# Search "alice" across all fields — should return both
|
||||||
|
result = await UserSearchCrud.offset_paginate(
|
||||||
|
db_session, search="alice", schema=UserRead
|
||||||
|
)
|
||||||
|
|
||||||
|
assert isinstance(result.pagination, OffsetPagination)
|
||||||
|
assert result.pagination.total_count == 2
|
||||||
|
|
||||||
|
@pytest.mark.anyio
|
||||||
|
async def test_search_column_with_cursor_paginate(self, db_session: AsyncSession):
|
||||||
|
"""search_column works with cursor_paginate."""
|
||||||
|
UserSearchCursorCrud = CrudFactory(
|
||||||
|
User,
|
||||||
|
cursor_column=User.id,
|
||||||
|
searchable_fields=[User.username, User.email],
|
||||||
|
)
|
||||||
|
await UserCrud.create(
|
||||||
|
db_session, UserCreate(username="alice", email="bob@test.com")
|
||||||
|
)
|
||||||
|
await UserCrud.create(
|
||||||
|
db_session, UserCreate(username="bob", email="alice@test.com")
|
||||||
|
)
|
||||||
|
|
||||||
|
result = await UserSearchCursorCrud.cursor_paginate(
|
||||||
|
db_session, search="alice", search_column="email", schema=UserRead
|
||||||
|
)
|
||||||
|
|
||||||
|
assert len(result.data) == 1
|
||||||
|
assert result.data[0].username == "bob"
|
||||||
|
|
||||||
|
|
||||||
|
class TestOrderParamsViaConsolidated:
|
||||||
|
"""Tests for order params via consolidated offset_paginate_params()."""
|
||||||
|
|
||||||
def test_generates_order_by_and_order_params(self):
|
def test_generates_order_by_and_order_params(self):
|
||||||
"""Returned dependency has order_by and order query params."""
|
"""Returned dependency has order_by and order query params."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
||||||
dep = UserOrderCrud.order_params()
|
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
|
||||||
|
|
||||||
param_names = set(inspect.signature(dep).parameters)
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
assert param_names == {"order_by", "order"}
|
assert "order_by" in param_names
|
||||||
|
assert "order" in param_names
|
||||||
|
|
||||||
def test_dependency_name_includes_model_name(self):
|
def test_order_disabled_no_order_params(self):
|
||||||
"""Dependency function is named after the model."""
|
"""When order=False, no order params are generated."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||||
dep = UserOrderCrud.order_params()
|
dep = UserOrderCrud.offset_paginate_params(
|
||||||
assert getattr(dep, "__name__") == "UserOrderParams"
|
search=False, filter=False, order=False
|
||||||
|
)
|
||||||
|
|
||||||
def test_raises_when_no_order_fields(self):
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
"""ValueError raised when no order_fields are configured or provided."""
|
assert "order_by" not in param_names
|
||||||
with pytest.raises(ValueError, match="no order_fields"):
|
assert "order" not in param_names
|
||||||
UserCrud.order_params()
|
|
||||||
|
|
||||||
def test_order_fields_override(self):
|
def test_order_fields_override(self):
|
||||||
"""order_fields= parameter overrides the class-level default."""
|
"""order_fields= parameter overrides the class-level default."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
||||||
dep = UserOrderCrud.order_params(order_fields=[User.email])
|
dep = UserOrderCrud.offset_paginate_params(
|
||||||
|
search=False, filter=False, order_fields=[User.email]
|
||||||
|
)
|
||||||
|
|
||||||
param_names = set(inspect.signature(dep).parameters)
|
|
||||||
assert "order_by" in param_names
|
|
||||||
# description should only mention email, not username
|
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
description = sig.parameters["order_by"].default.description
|
description = sig.parameters["order_by"].default.description
|
||||||
assert "email" in description
|
assert "email" in description
|
||||||
@@ -1237,7 +1464,7 @@ class TestOrderParamsSchema:
|
|||||||
def test_order_by_description_lists_valid_fields(self):
|
def test_order_by_description_lists_valid_fields(self):
|
||||||
"""order_by query param description mentions each allowed field."""
|
"""order_by query param description mentions each allowed field."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
||||||
dep = UserOrderCrud.order_params()
|
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
|
||||||
|
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
description = sig.parameters["order_by"].default.description
|
description = sig.parameters["order_by"].default.description
|
||||||
@@ -1247,8 +1474,12 @@ class TestOrderParamsSchema:
|
|||||||
def test_default_order_reflected_in_order_default(self):
|
def test_default_order_reflected_in_order_default(self):
|
||||||
"""default_order is used as the default value for order."""
|
"""default_order is used as the default value for order."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||||
dep_asc = UserOrderCrud.order_params(default_order="asc")
|
dep_asc = UserOrderCrud.offset_paginate_params(
|
||||||
dep_desc = UserOrderCrud.order_params(default_order="desc")
|
search=False, filter=False, default_order="asc"
|
||||||
|
)
|
||||||
|
dep_desc = UserOrderCrud.offset_paginate_params(
|
||||||
|
search=False, filter=False, default_order="desc"
|
||||||
|
)
|
||||||
|
|
||||||
sig_asc = inspect.signature(dep_asc)
|
sig_asc = inspect.signature(dep_asc)
|
||||||
sig_desc = inspect.signature(dep_desc)
|
sig_desc = inspect.signature(dep_desc)
|
||||||
@@ -1257,55 +1488,59 @@ class TestOrderParamsSchema:
|
|||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_no_order_by_no_default_returns_none(self):
|
async def test_no_order_by_no_default_returns_none(self):
|
||||||
"""Returns None when order_by is absent and no default_field is set."""
|
"""Returns order_by=None when order_by is absent and no default_order_field is set."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||||
dep = UserOrderCrud.order_params()
|
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
|
||||||
result = await dep(order_by=None, order="asc")
|
result = await dep(page=1, items_per_page=20, order_by=None, order="asc")
|
||||||
assert result is None
|
assert result["order_by"] is None
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_no_order_by_with_default_field_returns_asc_expression(self):
|
async def test_no_order_by_with_default_field_returns_asc_expression(self):
|
||||||
"""Returns default_field.asc() when order_by absent and order=asc."""
|
"""Returns default_order_field.asc() when order_by absent and order=asc."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||||
dep = UserOrderCrud.order_params(default_field=User.username)
|
dep = UserOrderCrud.offset_paginate_params(
|
||||||
result = await dep(order_by=None, order="asc")
|
search=False, filter=False, default_order_field=User.username
|
||||||
assert isinstance(result, UnaryExpression)
|
)
|
||||||
assert "ASC" in str(result)
|
result = await dep(page=1, items_per_page=20, order_by=None, order="asc")
|
||||||
|
assert isinstance(result["order_by"], UnaryExpression)
|
||||||
|
assert "ASC" in str(result["order_by"])
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_no_order_by_with_default_field_returns_desc_expression(self):
|
async def test_no_order_by_with_default_field_returns_desc_expression(self):
|
||||||
"""Returns default_field.desc() when order_by absent and order=desc."""
|
"""Returns default_order_field.desc() when order_by absent and order=desc."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||||
dep = UserOrderCrud.order_params(default_field=User.username)
|
dep = UserOrderCrud.offset_paginate_params(
|
||||||
result = await dep(order_by=None, order="desc")
|
search=False, filter=False, default_order_field=User.username
|
||||||
assert isinstance(result, UnaryExpression)
|
)
|
||||||
assert "DESC" in str(result)
|
result = await dep(page=1, items_per_page=20, order_by=None, order="desc")
|
||||||
|
assert isinstance(result["order_by"], UnaryExpression)
|
||||||
|
assert "DESC" in str(result["order_by"])
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_valid_order_by_asc(self):
|
async def test_valid_order_by_asc(self):
|
||||||
"""Returns field.asc() for a valid order_by with order=asc."""
|
"""Returns field.asc() for a valid order_by with order=asc."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||||
dep = UserOrderCrud.order_params()
|
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
|
||||||
result = await dep(order_by="username", order="asc")
|
result = await dep(page=1, items_per_page=20, order_by="username", order="asc")
|
||||||
assert isinstance(result, UnaryExpression)
|
assert isinstance(result["order_by"], UnaryExpression)
|
||||||
assert "ASC" in str(result)
|
assert "ASC" in str(result["order_by"])
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_valid_order_by_desc(self):
|
async def test_valid_order_by_desc(self):
|
||||||
"""Returns field.desc() for a valid order_by with order=desc."""
|
"""Returns field.desc() for a valid order_by with order=desc."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||||
dep = UserOrderCrud.order_params()
|
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
|
||||||
result = await dep(order_by="username", order="desc")
|
result = await dep(page=1, items_per_page=20, order_by="username", order="desc")
|
||||||
assert isinstance(result, UnaryExpression)
|
assert isinstance(result["order_by"], UnaryExpression)
|
||||||
assert "DESC" in str(result)
|
assert "DESC" in str(result["order_by"])
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_invalid_order_by_raises_invalid_order_field_error(self):
|
async def test_invalid_order_by_raises_invalid_order_field_error(self):
|
||||||
"""Raises InvalidOrderFieldError for an unknown order_by value."""
|
"""Raises InvalidOrderFieldError for an unknown order_by value."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||||
dep = UserOrderCrud.order_params()
|
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
|
||||||
with pytest.raises(InvalidOrderFieldError) as exc_info:
|
with pytest.raises(InvalidOrderFieldError) as exc_info:
|
||||||
await dep(order_by="nonexistent", order="asc")
|
await dep(page=1, items_per_page=20, order_by="nonexistent", order="asc")
|
||||||
assert exc_info.value.field == "nonexistent"
|
assert exc_info.value.field == "nonexistent"
|
||||||
assert "username" in exc_info.value.valid_fields
|
assert "username" in exc_info.value.valid_fields
|
||||||
|
|
||||||
@@ -1313,17 +1548,21 @@ class TestOrderParamsSchema:
|
|||||||
async def test_multiple_fields_all_resolve(self):
|
async def test_multiple_fields_all_resolve(self):
|
||||||
"""All configured fields resolve correctly via order_by."""
|
"""All configured fields resolve correctly via order_by."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
||||||
dep = UserOrderCrud.order_params()
|
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
|
||||||
result_username = await dep(order_by="username", order="asc")
|
result_username = await dep(
|
||||||
result_email = await dep(order_by="email", order="desc")
|
page=1, items_per_page=20, order_by="username", order="asc"
|
||||||
assert isinstance(result_username, ColumnElement)
|
)
|
||||||
assert isinstance(result_email, ColumnElement)
|
result_email = await dep(
|
||||||
|
page=1, items_per_page=20, order_by="email", order="desc"
|
||||||
|
)
|
||||||
|
assert isinstance(result_username["order_by"], ColumnElement)
|
||||||
|
assert isinstance(result_email["order_by"], ColumnElement)
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_order_params_integrates_with_get_multi(
|
async def test_order_integrates_with_offset_paginate(
|
||||||
self, db_session: AsyncSession
|
self, db_session: AsyncSession
|
||||||
):
|
):
|
||||||
"""order_params output is accepted by get_multi(order_by=...)."""
|
"""order in consolidated params is accepted by offset_paginate(order_by=...)."""
|
||||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||||
await UserCrud.create(
|
await UserCrud.create(
|
||||||
db_session, UserCreate(username="charlie", email="c@test.com")
|
db_session, UserCreate(username="charlie", email="c@test.com")
|
||||||
@@ -1332,37 +1571,43 @@ class TestOrderParamsSchema:
|
|||||||
db_session, UserCreate(username="alice", email="a@test.com")
|
db_session, UserCreate(username="alice", email="a@test.com")
|
||||||
)
|
)
|
||||||
|
|
||||||
dep = UserOrderCrud.order_params()
|
dep = UserOrderCrud.offset_paginate_params(search=False, filter=False)
|
||||||
order_by = await dep(order_by="username", order="asc")
|
params = await dep(page=1, items_per_page=20, order_by="username", order="asc")
|
||||||
results = await UserOrderCrud.get_multi(db_session, order_by=order_by)
|
result = await UserOrderCrud.offset_paginate(
|
||||||
|
db_session, **params, schema=UserRead
|
||||||
|
)
|
||||||
|
|
||||||
assert results[0].username == "alice"
|
assert result.data[0].username == "alice"
|
||||||
assert results[1].username == "charlie"
|
assert result.data[1].username == "charlie"
|
||||||
|
|
||||||
|
|
||||||
class TestOffsetParamsSchema:
|
class TestOffsetPaginateParamsSchema:
|
||||||
"""Tests for AsyncCrud.offset_params()."""
|
"""Tests for AsyncCrud.offset_paginate_params()."""
|
||||||
|
|
||||||
def test_returns_page_and_items_per_page_params(self):
|
def test_returns_page_and_items_per_page_params(self):
|
||||||
"""Returned dependency has page and items_per_page params only."""
|
"""Returned dependency has page and items_per_page params."""
|
||||||
dep = RoleCrud.offset_params()
|
dep = RoleCrud.offset_paginate_params(search=False, filter=False, order=False)
|
||||||
param_names = set(inspect.signature(dep).parameters)
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
assert param_names == {"page", "items_per_page"}
|
assert param_names == {"page", "items_per_page"}
|
||||||
|
|
||||||
def test_dependency_name_includes_model_name(self):
|
def test_dependency_name_includes_model_name(self):
|
||||||
"""Dependency function is named after the model."""
|
"""Dependency function is named after the model."""
|
||||||
dep = RoleCrud.offset_params()
|
dep = RoleCrud.offset_paginate_params(search=False, filter=False, order=False)
|
||||||
assert getattr(dep, "__name__") == "RoleOffsetParams"
|
assert getattr(dep, "__name__") == "RoleOffsetPaginateParams"
|
||||||
|
|
||||||
def test_default_page_size_reflected_in_items_per_page_default(self):
|
def test_default_page_size_reflected_in_items_per_page_default(self):
|
||||||
"""default_page_size is used as the default for items_per_page."""
|
"""default_page_size is used as the default for items_per_page."""
|
||||||
dep = RoleCrud.offset_params(default_page_size=42)
|
dep = RoleCrud.offset_paginate_params(
|
||||||
|
default_page_size=42, search=False, filter=False, order=False
|
||||||
|
)
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
assert sig.parameters["items_per_page"].default.default == 42
|
assert sig.parameters["items_per_page"].default.default == 42
|
||||||
|
|
||||||
def test_max_page_size_reflected_in_items_per_page_le(self):
|
def test_max_page_size_reflected_in_items_per_page_le(self):
|
||||||
"""max_page_size is used as le constraint on items_per_page."""
|
"""max_page_size is used as le constraint on items_per_page."""
|
||||||
dep = RoleCrud.offset_params(max_page_size=50)
|
dep = RoleCrud.offset_paginate_params(
|
||||||
|
max_page_size=50, search=False, filter=False, order=False
|
||||||
|
)
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
le = next(
|
le = next(
|
||||||
m.le
|
m.le
|
||||||
@@ -1373,67 +1618,121 @@ class TestOffsetParamsSchema:
|
|||||||
|
|
||||||
def test_include_total_not_a_query_param(self):
|
def test_include_total_not_a_query_param(self):
|
||||||
"""include_total is not exposed as a query parameter."""
|
"""include_total is not exposed as a query parameter."""
|
||||||
dep = RoleCrud.offset_params()
|
dep = RoleCrud.offset_paginate_params(search=False, filter=False, order=False)
|
||||||
param_names = set(inspect.signature(dep).parameters)
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
assert "include_total" not in param_names
|
assert "include_total" not in param_names
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_include_total_true_forwarded_in_result(self):
|
async def test_include_total_true_forwarded_in_result(self):
|
||||||
"""include_total=True factory arg appears in the resolved dict."""
|
"""include_total=True factory arg appears in the resolved dict."""
|
||||||
result = await RoleCrud.offset_params(include_total=True)(
|
result = await RoleCrud.offset_paginate_params(
|
||||||
page=1, items_per_page=10
|
include_total=True, search=False, filter=False, order=False
|
||||||
)
|
)(page=1, items_per_page=10)
|
||||||
assert result["include_total"] is True
|
assert result["include_total"] is True
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_include_total_false_forwarded_in_result(self):
|
async def test_include_total_false_forwarded_in_result(self):
|
||||||
"""include_total=False factory arg appears in the resolved dict."""
|
"""include_total=False factory arg appears in the resolved dict."""
|
||||||
result = await RoleCrud.offset_params(include_total=False)(
|
result = await RoleCrud.offset_paginate_params(
|
||||||
page=1, items_per_page=10
|
include_total=False, search=False, filter=False, order=False
|
||||||
)
|
)(page=1, items_per_page=10)
|
||||||
assert result["include_total"] is False
|
assert result["include_total"] is False
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_awaiting_dep_returns_dict(self):
|
async def test_awaiting_dep_returns_dict(self):
|
||||||
"""Awaiting the dependency returns a dict with page, items_per_page, include_total."""
|
"""Awaiting the dependency returns a dict with page, items_per_page, include_total."""
|
||||||
dep = RoleCrud.offset_params(include_total=False)
|
dep = RoleCrud.offset_paginate_params(
|
||||||
|
include_total=False, search=False, filter=False, order=False
|
||||||
|
)
|
||||||
result = await dep(page=2, items_per_page=10)
|
result = await dep(page=2, items_per_page=10)
|
||||||
assert result == {"page": 2, "items_per_page": 10, "include_total": False}
|
assert result == {"page": 2, "items_per_page": 10, "include_total": False}
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_integrates_with_offset_paginate(self, db_session: AsyncSession):
|
async def test_integrates_with_offset_paginate(self, db_session: AsyncSession):
|
||||||
"""offset_params output can be unpacked directly into offset_paginate."""
|
"""offset_paginate_params output can be unpacked directly into offset_paginate."""
|
||||||
await RoleCrud.create(db_session, RoleCreate(name="admin"))
|
await RoleCrud.create(db_session, RoleCreate(name="admin"))
|
||||||
dep = RoleCrud.offset_params()
|
dep = RoleCrud.offset_paginate_params(search=False, filter=False, order=False)
|
||||||
params = await dep(page=1, items_per_page=10)
|
params = await dep(page=1, items_per_page=10)
|
||||||
result = await RoleCrud.offset_paginate(db_session, **params, schema=RoleRead)
|
result = await RoleCrud.offset_paginate(db_session, **params, schema=RoleRead)
|
||||||
assert result.pagination.page == 1
|
assert result.pagination.page == 1
|
||||||
assert result.pagination.items_per_page == 10
|
assert result.pagination.items_per_page == 10
|
||||||
|
|
||||||
|
def test_all_features_enabled(self):
|
||||||
|
"""With all features enabled, params include search, filter, and order."""
|
||||||
|
FullCrud = CrudFactory(
|
||||||
|
User,
|
||||||
|
searchable_fields=[User.username],
|
||||||
|
facet_fields=[User.email],
|
||||||
|
order_fields=[User.username],
|
||||||
|
)
|
||||||
|
dep = FullCrud.offset_paginate_params()
|
||||||
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
|
assert param_names == {
|
||||||
|
"page",
|
||||||
|
"items_per_page",
|
||||||
|
"search",
|
||||||
|
"search_column",
|
||||||
|
"email",
|
||||||
|
"order_by",
|
||||||
|
"order",
|
||||||
|
}
|
||||||
|
|
||||||
class TestCursorParamsSchema:
|
def test_search_enabled_but_no_searchable_fields(self):
|
||||||
"""Tests for AsyncCrud.cursor_params()."""
|
"""search=True with no searchable_fields silently skips search params."""
|
||||||
|
NoCrud = CrudFactory(Role)
|
||||||
|
NoCrud.searchable_fields = None
|
||||||
|
dep = NoCrud.offset_paginate_params(
|
||||||
|
search=True, filter=False, order=False, search_fields=None
|
||||||
|
)
|
||||||
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
|
assert "search" not in param_names
|
||||||
|
assert "search_column" not in param_names
|
||||||
|
|
||||||
|
def test_filter_enabled_but_no_facet_fields(self):
|
||||||
|
"""filter=True with no facet_fields silently skips filter params."""
|
||||||
|
dep = RoleCrud.offset_paginate_params(search=False, filter=True, order=False)
|
||||||
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
|
assert param_names == {"page", "items_per_page"}
|
||||||
|
|
||||||
|
def test_order_enabled_but_no_order_fields(self):
|
||||||
|
"""order=True with no order_fields silently skips order params."""
|
||||||
|
dep = RoleCrud.offset_paginate_params(search=False, filter=False, order=True)
|
||||||
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
|
assert "order_by" not in param_names
|
||||||
|
assert "order" not in param_names
|
||||||
|
|
||||||
|
|
||||||
|
class TestCursorPaginateParamsSchema:
|
||||||
|
"""Tests for AsyncCrud.cursor_paginate_params()."""
|
||||||
|
|
||||||
def test_returns_cursor_and_items_per_page_params(self):
|
def test_returns_cursor_and_items_per_page_params(self):
|
||||||
"""Returned dependency has cursor and items_per_page params."""
|
"""Returned dependency has cursor and items_per_page params."""
|
||||||
dep = RoleCursorCrud.cursor_params()
|
dep = RoleCursorCrud.cursor_paginate_params(
|
||||||
|
search=False, filter=False, order=False
|
||||||
|
)
|
||||||
param_names = set(inspect.signature(dep).parameters)
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
assert param_names == {"cursor", "items_per_page"}
|
assert param_names == {"cursor", "items_per_page"}
|
||||||
|
|
||||||
def test_dependency_name_includes_model_name(self):
|
def test_dependency_name_includes_model_name(self):
|
||||||
"""Dependency function is named after the model."""
|
"""Dependency function is named after the model."""
|
||||||
dep = RoleCursorCrud.cursor_params()
|
dep = RoleCursorCrud.cursor_paginate_params(
|
||||||
assert getattr(dep, "__name__") == "RoleCursorParams"
|
search=False, filter=False, order=False
|
||||||
|
)
|
||||||
|
assert getattr(dep, "__name__") == "RoleCursorPaginateParams"
|
||||||
|
|
||||||
def test_default_page_size_reflected_in_items_per_page_default(self):
|
def test_default_page_size_reflected_in_items_per_page_default(self):
|
||||||
"""default_page_size is used as the default for items_per_page."""
|
"""default_page_size is used as the default for items_per_page."""
|
||||||
dep = RoleCursorCrud.cursor_params(default_page_size=15)
|
dep = RoleCursorCrud.cursor_paginate_params(
|
||||||
|
default_page_size=15, search=False, filter=False, order=False
|
||||||
|
)
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
assert sig.parameters["items_per_page"].default.default == 15
|
assert sig.parameters["items_per_page"].default.default == 15
|
||||||
|
|
||||||
def test_max_page_size_reflected_in_items_per_page_le(self):
|
def test_max_page_size_reflected_in_items_per_page_le(self):
|
||||||
"""max_page_size is used as le constraint on items_per_page."""
|
"""max_page_size is used as le constraint on items_per_page."""
|
||||||
dep = RoleCursorCrud.cursor_params(max_page_size=75)
|
dep = RoleCursorCrud.cursor_paginate_params(
|
||||||
|
max_page_size=75, search=False, filter=False, order=False
|
||||||
|
)
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
le = next(
|
le = next(
|
||||||
m.le
|
m.le
|
||||||
@@ -1444,22 +1743,28 @@ class TestCursorParamsSchema:
|
|||||||
|
|
||||||
def test_cursor_defaults_to_none(self):
|
def test_cursor_defaults_to_none(self):
|
||||||
"""cursor defaults to None."""
|
"""cursor defaults to None."""
|
||||||
dep = RoleCursorCrud.cursor_params()
|
dep = RoleCursorCrud.cursor_paginate_params(
|
||||||
|
search=False, filter=False, order=False
|
||||||
|
)
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
assert sig.parameters["cursor"].default.default is None
|
assert sig.parameters["cursor"].default.default is None
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_awaiting_dep_returns_dict(self):
|
async def test_awaiting_dep_returns_dict(self):
|
||||||
"""Awaiting the dependency returns a dict with cursor and items_per_page."""
|
"""Awaiting the dependency returns a dict with cursor and items_per_page."""
|
||||||
dep = RoleCursorCrud.cursor_params()
|
dep = RoleCursorCrud.cursor_paginate_params(
|
||||||
|
search=False, filter=False, order=False
|
||||||
|
)
|
||||||
result = await dep(cursor=None, items_per_page=5)
|
result = await dep(cursor=None, items_per_page=5)
|
||||||
assert result == {"cursor": None, "items_per_page": 5}
|
assert result == {"cursor": None, "items_per_page": 5}
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_integrates_with_cursor_paginate(self, db_session: AsyncSession):
|
async def test_integrates_with_cursor_paginate(self, db_session: AsyncSession):
|
||||||
"""cursor_params output can be unpacked directly into cursor_paginate."""
|
"""cursor_paginate_params output can be unpacked directly into cursor_paginate."""
|
||||||
await RoleCrud.create(db_session, RoleCreate(name="admin"))
|
await RoleCrud.create(db_session, RoleCreate(name="admin"))
|
||||||
dep = RoleCursorCrud.cursor_params()
|
dep = RoleCursorCrud.cursor_paginate_params(
|
||||||
|
search=False, filter=False, order=False
|
||||||
|
)
|
||||||
params = await dep(cursor=None, items_per_page=10)
|
params = await dep(cursor=None, items_per_page=10)
|
||||||
result = await RoleCursorCrud.cursor_paginate(
|
result = await RoleCursorCrud.cursor_paginate(
|
||||||
db_session, **params, schema=RoleRead
|
db_session, **params, schema=RoleRead
|
||||||
@@ -1472,13 +1777,13 @@ class TestPaginateParamsSchema:
|
|||||||
|
|
||||||
def test_returns_all_params(self):
|
def test_returns_all_params(self):
|
||||||
"""Returned dependency has pagination_type, page, cursor, items_per_page (no include_total)."""
|
"""Returned dependency has pagination_type, page, cursor, items_per_page (no include_total)."""
|
||||||
dep = RoleCursorCrud.paginate_params()
|
dep = RoleCursorCrud.paginate_params(search=False, filter=False, order=False)
|
||||||
param_names = set(inspect.signature(dep).parameters)
|
param_names = set(inspect.signature(dep).parameters)
|
||||||
assert param_names == {"pagination_type", "page", "cursor", "items_per_page"}
|
assert param_names == {"pagination_type", "page", "cursor", "items_per_page"}
|
||||||
|
|
||||||
def test_dependency_name_includes_model_name(self):
|
def test_dependency_name_includes_model_name(self):
|
||||||
"""Dependency function is named after the model."""
|
"""Dependency function is named after the model."""
|
||||||
dep = RoleCursorCrud.paginate_params()
|
dep = RoleCursorCrud.paginate_params(search=False, filter=False, order=False)
|
||||||
assert getattr(dep, "__name__") == "RolePaginateParams"
|
assert getattr(dep, "__name__") == "RolePaginateParams"
|
||||||
|
|
||||||
def test_default_pagination_type(self):
|
def test_default_pagination_type(self):
|
||||||
@@ -1486,7 +1791,10 @@ class TestPaginateParamsSchema:
|
|||||||
from fastapi_toolsets.schemas import PaginationType
|
from fastapi_toolsets.schemas import PaginationType
|
||||||
|
|
||||||
dep = RoleCursorCrud.paginate_params(
|
dep = RoleCursorCrud.paginate_params(
|
||||||
default_pagination_type=PaginationType.CURSOR
|
default_pagination_type=PaginationType.CURSOR,
|
||||||
|
search=False,
|
||||||
|
filter=False,
|
||||||
|
order=False,
|
||||||
)
|
)
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
assert (
|
assert (
|
||||||
@@ -1495,13 +1803,17 @@ class TestPaginateParamsSchema:
|
|||||||
|
|
||||||
def test_default_page_size(self):
|
def test_default_page_size(self):
|
||||||
"""default_page_size is reflected in items_per_page default."""
|
"""default_page_size is reflected in items_per_page default."""
|
||||||
dep = RoleCursorCrud.paginate_params(default_page_size=15)
|
dep = RoleCursorCrud.paginate_params(
|
||||||
|
default_page_size=15, search=False, filter=False, order=False
|
||||||
|
)
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
assert sig.parameters["items_per_page"].default.default == 15
|
assert sig.parameters["items_per_page"].default.default == 15
|
||||||
|
|
||||||
def test_max_page_size_le_constraint(self):
|
def test_max_page_size_le_constraint(self):
|
||||||
"""max_page_size is used as le constraint on items_per_page."""
|
"""max_page_size is used as le constraint on items_per_page."""
|
||||||
dep = RoleCursorCrud.paginate_params(max_page_size=60)
|
dep = RoleCursorCrud.paginate_params(
|
||||||
|
max_page_size=60, search=False, filter=False, order=False
|
||||||
|
)
|
||||||
sig = inspect.signature(dep)
|
sig = inspect.signature(dep)
|
||||||
le = next(
|
le = next(
|
||||||
m.le
|
m.le
|
||||||
@@ -1512,19 +1824,23 @@ class TestPaginateParamsSchema:
|
|||||||
|
|
||||||
def test_include_total_not_a_query_param(self):
|
def test_include_total_not_a_query_param(self):
|
||||||
"""include_total is not exposed as a query parameter."""
|
"""include_total is not exposed as a query parameter."""
|
||||||
dep = RoleCursorCrud.paginate_params()
|
dep = RoleCursorCrud.paginate_params(search=False, filter=False, order=False)
|
||||||
assert "include_total" not in set(inspect.signature(dep).parameters)
|
assert "include_total" not in set(inspect.signature(dep).parameters)
|
||||||
|
|
||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_include_total_forwarded_in_result(self):
|
async def test_include_total_forwarded_in_result(self):
|
||||||
"""include_total factory arg appears in the resolved dict."""
|
"""include_total factory arg appears in the resolved dict."""
|
||||||
result_true = await RoleCursorCrud.paginate_params(include_total=True)(
|
result_true = await RoleCursorCrud.paginate_params(
|
||||||
|
include_total=True, search=False, filter=False, order=False
|
||||||
|
)(
|
||||||
pagination_type=PaginationType.OFFSET,
|
pagination_type=PaginationType.OFFSET,
|
||||||
page=1,
|
page=1,
|
||||||
cursor=None,
|
cursor=None,
|
||||||
items_per_page=10,
|
items_per_page=10,
|
||||||
)
|
)
|
||||||
result_false = await RoleCursorCrud.paginate_params(include_total=False)(
|
result_false = await RoleCursorCrud.paginate_params(
|
||||||
|
include_total=False, search=False, filter=False, order=False
|
||||||
|
)(
|
||||||
pagination_type=PaginationType.OFFSET,
|
pagination_type=PaginationType.OFFSET,
|
||||||
page=1,
|
page=1,
|
||||||
cursor=None,
|
cursor=None,
|
||||||
@@ -1536,7 +1852,7 @@ class TestPaginateParamsSchema:
|
|||||||
@pytest.mark.anyio
|
@pytest.mark.anyio
|
||||||
async def test_awaiting_dep_returns_dict(self):
|
async def test_awaiting_dep_returns_dict(self):
|
||||||
"""Awaiting the dependency returns a dict with all pagination keys."""
|
"""Awaiting the dependency returns a dict with all pagination keys."""
|
||||||
dep = RoleCursorCrud.paginate_params()
|
dep = RoleCursorCrud.paginate_params(search=False, filter=False, order=False)
|
||||||
result = await dep(
|
result = await dep(
|
||||||
pagination_type=PaginationType.OFFSET,
|
pagination_type=PaginationType.OFFSET,
|
||||||
page=2,
|
page=2,
|
||||||
@@ -1557,7 +1873,9 @@ class TestPaginateParamsSchema:
|
|||||||
from fastapi_toolsets.schemas import OffsetPagination
|
from fastapi_toolsets.schemas import OffsetPagination
|
||||||
|
|
||||||
await RoleCrud.create(db_session, RoleCreate(name="admin"))
|
await RoleCrud.create(db_session, RoleCreate(name="admin"))
|
||||||
params = await RoleCursorCrud.paginate_params()(
|
params = await RoleCursorCrud.paginate_params(
|
||||||
|
search=False, filter=False, order=False
|
||||||
|
)(
|
||||||
pagination_type=PaginationType.OFFSET,
|
pagination_type=PaginationType.OFFSET,
|
||||||
page=1,
|
page=1,
|
||||||
cursor=None,
|
cursor=None,
|
||||||
@@ -1572,7 +1890,9 @@ class TestPaginateParamsSchema:
|
|||||||
from fastapi_toolsets.schemas import CursorPagination
|
from fastapi_toolsets.schemas import CursorPagination
|
||||||
|
|
||||||
await RoleCrud.create(db_session, RoleCreate(name="admin"))
|
await RoleCrud.create(db_session, RoleCreate(name="admin"))
|
||||||
params = await RoleCursorCrud.paginate_params()(
|
params = await RoleCursorCrud.paginate_params(
|
||||||
|
search=False, filter=False, order=False
|
||||||
|
)(
|
||||||
pagination_type=PaginationType.CURSOR,
|
pagination_type=PaginationType.CURSOR,
|
||||||
page=1,
|
page=1,
|
||||||
cursor=None,
|
cursor=None,
|
||||||
|
|||||||
@@ -374,19 +374,6 @@ class TestCreateDbSession:
|
|||||||
pass
|
pass
|
||||||
|
|
||||||
|
|
||||||
class TestDeprecatedCleanupTables:
|
|
||||||
"""Tests for the deprecated cleanup_tables re-export in fastapi_toolsets.pytest."""
|
|
||||||
|
|
||||||
@pytest.mark.anyio
|
|
||||||
async def test_emits_deprecation_warning(self):
|
|
||||||
"""cleanup_tables imported from fastapi_toolsets.pytest emits DeprecationWarning."""
|
|
||||||
from fastapi_toolsets.pytest.utils import cleanup_tables
|
|
||||||
|
|
||||||
async with create_db_session(DATABASE_URL, Base, drop_tables=True) as session:
|
|
||||||
with pytest.warns(DeprecationWarning, match="fastapi_toolsets.db"):
|
|
||||||
await cleanup_tables(session, Base)
|
|
||||||
|
|
||||||
|
|
||||||
class TestGetXdistWorker:
|
class TestGetXdistWorker:
|
||||||
"""Tests for _get_xdist_worker helper."""
|
"""Tests for _get_xdist_worker helper."""
|
||||||
|
|
||||||
|
|||||||
6
uv.lock
generated
6
uv.lock
generated
@@ -894,15 +894,15 @@ wheels = [
|
|||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "pymdown-extensions"
|
name = "pymdown-extensions"
|
||||||
version = "10.21"
|
version = "10.21.2"
|
||||||
source = { registry = "https://pypi.org/simple" }
|
source = { registry = "https://pypi.org/simple" }
|
||||||
dependencies = [
|
dependencies = [
|
||||||
{ name = "markdown" },
|
{ name = "markdown" },
|
||||||
{ name = "pyyaml" },
|
{ name = "pyyaml" },
|
||||||
]
|
]
|
||||||
sdist = { url = "https://files.pythonhosted.org/packages/ba/63/06673d1eb6d8f83c0ea1f677d770e12565fb516928b4109c9e2055656a9e/pymdown_extensions-10.21.tar.gz", hash = "sha256:39f4a020f40773f6b2ff31d2cd2546c2c04d0a6498c31d9c688d2be07e1767d5", size = 853363, upload-time = "2026-02-15T20:44:06.748Z" }
|
sdist = { url = "https://files.pythonhosted.org/packages/df/08/f1c908c581fd11913da4711ea7ba32c0eee40b0190000996bb863b0c9349/pymdown_extensions-10.21.2.tar.gz", hash = "sha256:c3f55a5b8a1d0edf6699e35dcbea71d978d34ff3fa79f3d807b8a5b3fa90fbdc", size = 853922, upload-time = "2026-03-29T15:01:55.233Z" }
|
||||||
wheels = [
|
wheels = [
|
||||||
{ url = "https://files.pythonhosted.org/packages/6f/2c/5b079febdc65e1c3fb2729bf958d18b45be7113828528e8a0b5850dd819a/pymdown_extensions-10.21-py3-none-any.whl", hash = "sha256:91b879f9f864d49794c2d9534372b10150e6141096c3908a455e45ca72ad9d3f", size = 268877, upload-time = "2026-02-15T20:44:05.464Z" },
|
{ url = "https://files.pythonhosted.org/packages/f7/27/a2fc51a4a122dfd1015e921ae9d22fee3d20b0b8080d9a704578bf9deece/pymdown_extensions-10.21.2-py3-none-any.whl", hash = "sha256:5c0fd2a2bea14eb39af8ff284f1066d898ab2187d81b889b75d46d4348c01638", size = 268901, upload-time = "2026-03-29T15:01:53.244Z" },
|
||||||
]
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
|
|||||||
Reference in New Issue
Block a user