mirror of
https://github.com/d3vyce/fastapi-toolsets.git
synced 2026-03-01 17:00:48 +01:00
Compare commits
2 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
56d365d14b
|
|||
|
|
a257d85d45 |
@@ -1,6 +1,6 @@
|
||||
# Pagination & search
|
||||
|
||||
This example builds an articles listing endpoint that supports **offset pagination**, **cursor pagination**, **full-text search**, and **faceted filtering** — all from a single `CrudFactory` definition.
|
||||
This example builds an articles listing endpoint that supports **offset pagination**, **cursor pagination**, **full-text search**, **faceted filtering**, and **sorting** — all from a single `CrudFactory` definition.
|
||||
|
||||
## Models
|
||||
|
||||
@@ -16,7 +16,7 @@ This example builds an articles listing endpoint that supports **offset paginati
|
||||
|
||||
## Crud
|
||||
|
||||
Declare `facet_fields` and `searchable_fields` once on [`CrudFactory`](../reference/crud.md#fastapi_toolsets.crud.factory.CrudFactory). All endpoints built from this class share the same defaults and can override them per call.
|
||||
Declare `searchable_fields`, `facet_fields`, and `order_fields` once on [`CrudFactory`](../reference/crud.md#fastapi_toolsets.crud.factory.CrudFactory). All endpoints built from this class share the same defaults and can override them per call.
|
||||
|
||||
```python title="crud.py"
|
||||
--8<-- "docs_src/examples/pagination_search/crud.py"
|
||||
@@ -46,14 +46,14 @@ Declare `facet_fields` and `searchable_fields` once on [`CrudFactory`](../refere
|
||||
|
||||
Best for admin panels or any UI that needs a total item count and numbered pages.
|
||||
|
||||
```python title="routes.py:1:27"
|
||||
--8<-- "docs_src/examples/pagination_search/routes.py:1:27"
|
||||
```python title="routes.py:1:36"
|
||||
--8<-- "docs_src/examples/pagination_search/routes.py:1:36"
|
||||
```
|
||||
|
||||
**Example request**
|
||||
|
||||
```
|
||||
GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published
|
||||
GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published&order_by=title&order=asc
|
||||
```
|
||||
|
||||
**Example response**
|
||||
@@ -83,14 +83,14 @@ GET /articles/offset?page=2&items_per_page=10&search=fastapi&status=published
|
||||
|
||||
Best for feeds, infinite scroll, or any high-throughput API where offset performance degrades.
|
||||
|
||||
```python title="routes.py:30:45"
|
||||
--8<-- "docs_src/examples/pagination_search/routes.py:30:45"
|
||||
```python title="routes.py:39:59"
|
||||
--8<-- "docs_src/examples/pagination_search/routes.py:39:59"
|
||||
```
|
||||
|
||||
**Example request**
|
||||
|
||||
```
|
||||
GET /articles/cursor?items_per_page=10&status=published
|
||||
GET /articles/cursor?items_per_page=10&status=published&order_by=created_at&order=desc
|
||||
```
|
||||
|
||||
**Example response**
|
||||
|
||||
@@ -295,6 +295,8 @@ Use `filter_by` to pass the client's chosen filter values directly — no need t
|
||||
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:
|
||||
|
||||
```python
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends
|
||||
|
||||
UserCrud = CrudFactory(
|
||||
@@ -306,7 +308,7 @@ UserCrud = CrudFactory(
|
||||
async def list_users(
|
||||
session: SessionDep,
|
||||
page: int = 1,
|
||||
filter_by: dict[str, list[str]] = Depends(UserCrud.filter_params()),
|
||||
filter_by: Annotated[dict[str, list[str]], Depends(UserCrud.filter_params())],
|
||||
) -> PaginatedResponse[UserRead]:
|
||||
return await UserCrud.offset_paginate(
|
||||
session=session,
|
||||
@@ -323,6 +325,58 @@ GET /users?status=active&country=FR → filter_by={"status": ["active"], "coun
|
||||
GET /users?role=admin&role=editor → filter_by={"role": ["admin", "editor"]} (IN clause)
|
||||
```
|
||||
|
||||
## Sorting
|
||||
|
||||
!!! info "Added in `v1.3`"
|
||||
|
||||
Declare `order_fields` on the CRUD class to expose client-driven column ordering via `order_by` and `order` query parameters.
|
||||
|
||||
```python
|
||||
UserCrud = CrudFactory(
|
||||
model=User,
|
||||
order_fields=[
|
||||
User.name,
|
||||
User.created_at,
|
||||
],
|
||||
)
|
||||
```
|
||||
|
||||
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:
|
||||
|
||||
```python
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import Depends
|
||||
from fastapi_toolsets.crud import OrderByClause
|
||||
|
||||
@router.get("")
|
||||
async def list_users(
|
||||
session: SessionDep,
|
||||
order_by: Annotated[OrderByClause | None, Depends(UserCrud.order_params())],
|
||||
) -> PaginatedResponse[UserRead]:
|
||||
return await UserCrud.offset_paginate(session=session, order_by=order_by)
|
||||
```
|
||||
|
||||
The dependency adds two query parameters to the endpoint:
|
||||
|
||||
| Parameter | Type |
|
||||
| ---------- | --------------- |
|
||||
| `order_by` | `str | null` |
|
||||
| `order` | `asc` or `desc` |
|
||||
|
||||
```
|
||||
GET /users?order_by=name&order=asc → ORDER BY users.name ASC
|
||||
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).
|
||||
|
||||
You can also pass `order_fields` directly to `order_params()` to override the class-level defaults without modifying them:
|
||||
|
||||
```python
|
||||
UserOrderParams = UserCrud.order_params(order_fields=[User.name])
|
||||
```
|
||||
|
||||
## Relationship loading
|
||||
|
||||
!!! info "Added in `v1.1`"
|
||||
|
||||
@@ -13,6 +13,7 @@ from fastapi_toolsets.exceptions import (
|
||||
ConflictError,
|
||||
NoSearchableFieldsError,
|
||||
InvalidFacetFilterError,
|
||||
InvalidOrderFieldError,
|
||||
generate_error_responses,
|
||||
init_exceptions_handlers,
|
||||
)
|
||||
@@ -32,6 +33,8 @@ from fastapi_toolsets.exceptions import (
|
||||
|
||||
## ::: fastapi_toolsets.exceptions.exceptions.InvalidFacetFilterError
|
||||
|
||||
## ::: fastapi_toolsets.exceptions.exceptions.InvalidOrderFieldError
|
||||
|
||||
## ::: fastapi_toolsets.exceptions.exceptions.generate_error_responses
|
||||
|
||||
## ::: fastapi_toolsets.exceptions.handler.init_exceptions_handlers
|
||||
|
||||
@@ -1,6 +1,9 @@
|
||||
from fastapi import FastAPI
|
||||
|
||||
from fastapi_toolsets.exceptions import init_exceptions_handlers
|
||||
|
||||
from .routes import router
|
||||
|
||||
app = FastAPI()
|
||||
init_exceptions_handlers(app=app)
|
||||
app.include_router(router=router)
|
||||
|
||||
@@ -14,6 +14,8 @@ ArticleCrud = CrudFactory(
|
||||
Article.status,
|
||||
(Article.category, Category.name),
|
||||
],
|
||||
order_fields=[ # fields exposed for client-driven ordering
|
||||
Article.title,
|
||||
Article.created_at,
|
||||
],
|
||||
)
|
||||
|
||||
ArticleFilters = ArticleCrud.filter_params()
|
||||
|
||||
@@ -1,9 +1,13 @@
|
||||
from typing import Annotated
|
||||
|
||||
from fastapi import APIRouter, Depends, Query
|
||||
|
||||
from fastapi_toolsets.crud import OrderByClause
|
||||
from fastapi_toolsets.schemas import PaginatedResponse
|
||||
|
||||
from .crud import ArticleCrud
|
||||
from .db import SessionDep
|
||||
from .models import Article
|
||||
from .schemas import ArticleRead
|
||||
|
||||
router = APIRouter(prefix="/articles")
|
||||
@@ -12,10 +16,14 @@ router = APIRouter(prefix="/articles")
|
||||
@router.get("/offset")
|
||||
async def list_articles_offset(
|
||||
session: SessionDep,
|
||||
filter_by: Annotated[dict[str, list[str]], Depends(ArticleCrud.filter_params())],
|
||||
order_by: Annotated[
|
||||
OrderByClause | None,
|
||||
Depends(ArticleCrud.order_params(default_field=Article.created_at)),
|
||||
],
|
||||
page: int = Query(1, ge=1),
|
||||
items_per_page: int = Query(20, ge=1, le=100),
|
||||
search: str | None = None,
|
||||
filter_by: dict[str, list[str]] = Depends(ArticleCrud.filter_params()),
|
||||
) -> PaginatedResponse[ArticleRead]:
|
||||
return await ArticleCrud.offset_paginate(
|
||||
session=session,
|
||||
@@ -23,6 +31,7 @@ async def list_articles_offset(
|
||||
items_per_page=items_per_page,
|
||||
search=search,
|
||||
filter_by=filter_by or None,
|
||||
order_by=order_by,
|
||||
schema=ArticleRead,
|
||||
)
|
||||
|
||||
@@ -30,10 +39,14 @@ async def list_articles_offset(
|
||||
@router.get("/cursor")
|
||||
async def list_articles_cursor(
|
||||
session: SessionDep,
|
||||
filter_by: Annotated[dict[str, list[str]], Depends(ArticleCrud.filter_params())],
|
||||
order_by: Annotated[
|
||||
OrderByClause | None,
|
||||
Depends(ArticleCrud.order_params(default_field=Article.created_at)),
|
||||
],
|
||||
cursor: str | None = None,
|
||||
items_per_page: int = Query(20, ge=1, le=100),
|
||||
search: str | None = None,
|
||||
filter_by: dict[str, list[str]] = Depends(ArticleCrud.filter_params()),
|
||||
) -> PaginatedResponse[ArticleRead]:
|
||||
return await ArticleCrud.cursor_paginate(
|
||||
session=session,
|
||||
@@ -41,5 +54,6 @@ async def list_articles_cursor(
|
||||
items_per_page=items_per_page,
|
||||
search=search,
|
||||
filter_by=filter_by or None,
|
||||
order_by=order_by,
|
||||
schema=ArticleRead,
|
||||
)
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "fastapi-toolsets"
|
||||
version = "1.2.1"
|
||||
version = "1.3.0"
|
||||
description = "Production-ready utilities for FastAPI applications"
|
||||
readme = "README.md"
|
||||
license = "MIT"
|
||||
|
||||
@@ -21,4 +21,4 @@ Example usage:
|
||||
return Response(data={"user": user.username}, message="Success")
|
||||
"""
|
||||
|
||||
__version__ = "1.2.1"
|
||||
__version__ = "1.3.0"
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
"""Generic async CRUD operations for SQLAlchemy models."""
|
||||
|
||||
from ..exceptions import InvalidFacetFilterError, NoSearchableFieldsError
|
||||
from .factory import CrudFactory, JoinType, M2MFieldType
|
||||
from .factory import CrudFactory, JoinType, M2MFieldType, OrderByClause
|
||||
from .search import (
|
||||
FacetFieldType,
|
||||
SearchConfig,
|
||||
@@ -16,5 +16,6 @@ __all__ = [
|
||||
"JoinType",
|
||||
"M2MFieldType",
|
||||
"NoSearchableFieldsError",
|
||||
"OrderByClause",
|
||||
"SearchConfig",
|
||||
]
|
||||
|
||||
@@ -21,10 +21,11 @@ from sqlalchemy.exc import NoResultFound
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.orm import DeclarativeBase, QueryableAttribute, selectinload
|
||||
from sqlalchemy.sql.base import ExecutableOption
|
||||
from sqlalchemy.sql.elements import ColumnElement
|
||||
from sqlalchemy.sql.roles import WhereHavingRole
|
||||
|
||||
from ..db import get_transaction
|
||||
from ..exceptions import NotFoundError
|
||||
from ..exceptions import InvalidOrderFieldError, NotFoundError
|
||||
from ..schemas import CursorPagination, OffsetPagination, PaginatedResponse, Response
|
||||
from .search import (
|
||||
FacetFieldType,
|
||||
@@ -40,6 +41,7 @@ ModelType = TypeVar("ModelType", bound=DeclarativeBase)
|
||||
SchemaType = TypeVar("SchemaType", bound=BaseModel)
|
||||
JoinType = list[tuple[type[DeclarativeBase], Any]]
|
||||
M2MFieldType = Mapping[str, QueryableAttribute[Any]]
|
||||
OrderByClause = ColumnElement[Any] | QueryableAttribute[Any]
|
||||
|
||||
|
||||
def _encode_cursor(value: Any) -> str:
|
||||
@@ -61,6 +63,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
model: ClassVar[type[DeclarativeBase]]
|
||||
searchable_fields: ClassVar[Sequence[SearchFieldType] | None] = None
|
||||
facet_fields: ClassVar[Sequence[FacetFieldType] | None] = None
|
||||
order_fields: ClassVar[Sequence[QueryableAttribute[Any]] | None] = None
|
||||
m2m_fields: ClassVar[M2MFieldType | None] = None
|
||||
default_load_options: ClassVar[list[ExecutableOption] | None] = None
|
||||
cursor_column: ClassVar[Any | None] = None
|
||||
@@ -176,6 +179,63 @@ class AsyncCrud(Generic[ModelType]):
|
||||
|
||||
return dependency
|
||||
|
||||
@classmethod
|
||||
def order_params(
|
||||
cls: type[Self],
|
||||
*,
|
||||
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
|
||||
default_field: QueryableAttribute[Any] | None = None,
|
||||
default_order: Literal["asc", "desc"] = "asc",
|
||||
) -> Callable[..., Awaitable[OrderByClause | None]]:
|
||||
"""Return a FastAPI dependency that resolves order query params into an order_by clause.
|
||||
|
||||
Args:
|
||||
order_fields: Override the allowed order fields. Falls back to the class-level
|
||||
``order_fields`` if not provided.
|
||||
default_field: Field to order by when ``order_by`` query param is absent.
|
||||
If ``None`` and no ``order_by`` is provided, no ordering is applied.
|
||||
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
|
||||
@classmethod
|
||||
async def create( # pragma: no cover
|
||||
@@ -415,7 +475,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
joins: JoinType | None = None,
|
||||
outer_join: bool = False,
|
||||
load_options: list[ExecutableOption] | None = None,
|
||||
order_by: Any | None = None,
|
||||
order_by: OrderByClause | None = None,
|
||||
limit: int | None = None,
|
||||
offset: int | None = None,
|
||||
) -> Sequence[ModelType]:
|
||||
@@ -745,7 +805,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
joins: JoinType | None = None,
|
||||
outer_join: bool = False,
|
||||
load_options: list[ExecutableOption] | None = None,
|
||||
order_by: Any | None = None,
|
||||
order_by: OrderByClause | None = None,
|
||||
page: int = 1,
|
||||
items_per_page: int = 20,
|
||||
search: str | SearchConfig | None = None,
|
||||
@@ -766,7 +826,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
joins: JoinType | None = None,
|
||||
outer_join: bool = False,
|
||||
load_options: list[ExecutableOption] | None = None,
|
||||
order_by: Any | None = None,
|
||||
order_by: OrderByClause | None = None,
|
||||
page: int = 1,
|
||||
items_per_page: int = 20,
|
||||
search: str | SearchConfig | None = None,
|
||||
@@ -785,7 +845,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
joins: JoinType | None = None,
|
||||
outer_join: bool = False,
|
||||
load_options: list[ExecutableOption] | None = None,
|
||||
order_by: Any | None = None,
|
||||
order_by: OrderByClause | None = None,
|
||||
page: int = 1,
|
||||
items_per_page: int = 20,
|
||||
search: str | SearchConfig | None = None,
|
||||
@@ -937,7 +997,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
joins: JoinType | None = None,
|
||||
outer_join: bool = False,
|
||||
load_options: list[ExecutableOption] | None = None,
|
||||
order_by: Any | None = None,
|
||||
order_by: OrderByClause | None = None,
|
||||
items_per_page: int = 20,
|
||||
search: str | SearchConfig | None = None,
|
||||
search_fields: Sequence[SearchFieldType] | None = None,
|
||||
@@ -958,7 +1018,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
joins: JoinType | None = None,
|
||||
outer_join: bool = False,
|
||||
load_options: list[ExecutableOption] | None = None,
|
||||
order_by: Any | None = None,
|
||||
order_by: OrderByClause | None = None,
|
||||
items_per_page: int = 20,
|
||||
search: str | SearchConfig | None = None,
|
||||
search_fields: Sequence[SearchFieldType] | None = None,
|
||||
@@ -977,7 +1037,7 @@ class AsyncCrud(Generic[ModelType]):
|
||||
joins: JoinType | None = None,
|
||||
outer_join: bool = False,
|
||||
load_options: list[ExecutableOption] | None = None,
|
||||
order_by: Any | None = None,
|
||||
order_by: OrderByClause | None = None,
|
||||
items_per_page: int = 20,
|
||||
search: str | SearchConfig | None = None,
|
||||
search_fields: Sequence[SearchFieldType] | None = None,
|
||||
@@ -1147,6 +1207,7 @@ def CrudFactory(
|
||||
*,
|
||||
searchable_fields: Sequence[SearchFieldType] | None = None,
|
||||
facet_fields: Sequence[FacetFieldType] | None = None,
|
||||
order_fields: Sequence[QueryableAttribute[Any]] | None = None,
|
||||
m2m_fields: M2MFieldType | None = None,
|
||||
default_load_options: list[ExecutableOption] | None = None,
|
||||
cursor_column: Any | None = None,
|
||||
@@ -1159,6 +1220,8 @@ def CrudFactory(
|
||||
facet_fields: Optional list of columns to compute distinct values for in paginated
|
||||
responses. Supports direct columns (``User.status``) and relationship tuples
|
||||
(``(User.role, Role.name)``). Can be overridden per call.
|
||||
order_fields: Optional list of model attributes that callers are allowed to order by
|
||||
via ``order_params()``. Can be overridden per call.
|
||||
m2m_fields: Optional mapping for many-to-many relationships.
|
||||
Maps schema field names (containing lists of IDs) to
|
||||
SQLAlchemy relationship attributes.
|
||||
@@ -1252,6 +1315,7 @@ def CrudFactory(
|
||||
"model": model,
|
||||
"searchable_fields": searchable_fields,
|
||||
"facet_fields": facet_fields,
|
||||
"order_fields": order_fields,
|
||||
"m2m_fields": m2m_fields,
|
||||
"default_load_options": default_load_options,
|
||||
"cursor_column": cursor_column,
|
||||
|
||||
@@ -6,6 +6,7 @@ from .exceptions import (
|
||||
ConflictError,
|
||||
ForbiddenError,
|
||||
InvalidFacetFilterError,
|
||||
InvalidOrderFieldError,
|
||||
NoSearchableFieldsError,
|
||||
NotFoundError,
|
||||
UnauthorizedError,
|
||||
@@ -21,6 +22,7 @@ __all__ = [
|
||||
"generate_error_responses",
|
||||
"init_exceptions_handlers",
|
||||
"InvalidFacetFilterError",
|
||||
"InvalidOrderFieldError",
|
||||
"NoSearchableFieldsError",
|
||||
"NotFoundError",
|
||||
"UnauthorizedError",
|
||||
|
||||
@@ -128,6 +128,31 @@ class InvalidFacetFilterError(ApiException):
|
||||
super().__init__(detail)
|
||||
|
||||
|
||||
class InvalidOrderFieldError(ApiException):
|
||||
"""Raised when order_by contains a field not in the allowed order fields."""
|
||||
|
||||
api_error = ApiError(
|
||||
code=422,
|
||||
msg="Invalid Order Field",
|
||||
desc="The requested order field is not allowed for this resource.",
|
||||
err_code="SORT-422",
|
||||
)
|
||||
|
||||
def __init__(self, field: str, valid_fields: list[str]) -> None:
|
||||
"""Initialize the exception.
|
||||
|
||||
Args:
|
||||
field: The unknown order field provided by the caller
|
||||
valid_fields: List of valid field names
|
||||
"""
|
||||
self.field = field
|
||||
self.valid_fields = valid_fields
|
||||
detail = (
|
||||
f"'{field}' is not an allowed order field. Valid fields: {valid_fields}."
|
||||
)
|
||||
super().__init__(detail)
|
||||
|
||||
|
||||
def generate_error_responses(
|
||||
*errors: type[ApiException],
|
||||
) -> dict[int | str, dict[str, Any]]:
|
||||
|
||||
@@ -1,9 +1,11 @@
|
||||
"""Tests for CRUD search functionality."""
|
||||
|
||||
import inspect
|
||||
import uuid
|
||||
|
||||
import pytest
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
from sqlalchemy.sql.elements import ColumnElement, UnaryExpression
|
||||
|
||||
from fastapi_toolsets.crud import (
|
||||
CrudFactory,
|
||||
@@ -11,6 +13,7 @@ from fastapi_toolsets.crud import (
|
||||
SearchConfig,
|
||||
get_searchable_fields,
|
||||
)
|
||||
from fastapi_toolsets.exceptions import InvalidOrderFieldError
|
||||
from fastapi_toolsets.schemas import OffsetPagination
|
||||
|
||||
from .conftest import (
|
||||
@@ -1014,3 +1017,144 @@ class TestFilterParamsSchema:
|
||||
|
||||
assert isinstance(result.pagination, OffsetPagination)
|
||||
assert result.pagination.total_count == 2
|
||||
|
||||
|
||||
class TestOrderParamsSchema:
|
||||
"""Tests for AsyncCrud.order_params()."""
|
||||
|
||||
def test_generates_order_by_and_order_params(self):
|
||||
"""Returned dependency has order_by and order query params."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
||||
dep = UserOrderCrud.order_params()
|
||||
|
||||
param_names = set(inspect.signature(dep).parameters)
|
||||
assert param_names == {"order_by", "order"}
|
||||
|
||||
def test_dependency_name_includes_model_name(self):
|
||||
"""Dependency function is named after the model."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||
dep = UserOrderCrud.order_params()
|
||||
assert getattr(dep, "__name__") == "UserOrderParams"
|
||||
|
||||
def test_raises_when_no_order_fields(self):
|
||||
"""ValueError raised when no order_fields are configured or provided."""
|
||||
with pytest.raises(ValueError, match="no order_fields"):
|
||||
UserCrud.order_params()
|
||||
|
||||
def test_order_fields_override(self):
|
||||
"""order_fields= parameter overrides the class-level default."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
||||
dep = UserOrderCrud.order_params(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)
|
||||
description = sig.parameters["order_by"].default.description
|
||||
assert "email" in description
|
||||
assert "username" not in description
|
||||
|
||||
def test_order_by_description_lists_valid_fields(self):
|
||||
"""order_by query param description mentions each allowed field."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
||||
dep = UserOrderCrud.order_params()
|
||||
|
||||
sig = inspect.signature(dep)
|
||||
description = sig.parameters["order_by"].default.description
|
||||
assert "username" in description
|
||||
assert "email" in description
|
||||
|
||||
def test_default_order_reflected_in_order_default(self):
|
||||
"""default_order is used as the default value for order."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||
dep_asc = UserOrderCrud.order_params(default_order="asc")
|
||||
dep_desc = UserOrderCrud.order_params(default_order="desc")
|
||||
|
||||
sig_asc = inspect.signature(dep_asc)
|
||||
sig_desc = inspect.signature(dep_desc)
|
||||
assert sig_asc.parameters["order"].default.default == "asc"
|
||||
assert sig_desc.parameters["order"].default.default == "desc"
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_no_order_by_no_default_returns_none(self):
|
||||
"""Returns None when order_by is absent and no default_field is set."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||
dep = UserOrderCrud.order_params()
|
||||
result = await dep(order_by=None, order="asc")
|
||||
assert result is None
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_no_order_by_with_default_field_returns_asc_expression(self):
|
||||
"""Returns default_field.asc() when order_by absent and order=asc."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||
dep = UserOrderCrud.order_params(default_field=User.username)
|
||||
result = await dep(order_by=None, order="asc")
|
||||
assert isinstance(result, UnaryExpression)
|
||||
assert "ASC" in str(result)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_no_order_by_with_default_field_returns_desc_expression(self):
|
||||
"""Returns default_field.desc() when order_by absent and order=desc."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||
dep = UserOrderCrud.order_params(default_field=User.username)
|
||||
result = await dep(order_by=None, order="desc")
|
||||
assert isinstance(result, UnaryExpression)
|
||||
assert "DESC" in str(result)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_valid_order_by_asc(self):
|
||||
"""Returns field.asc() for a valid order_by with order=asc."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||
dep = UserOrderCrud.order_params()
|
||||
result = await dep(order_by="username", order="asc")
|
||||
assert isinstance(result, UnaryExpression)
|
||||
assert "ASC" in str(result)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_valid_order_by_desc(self):
|
||||
"""Returns field.desc() for a valid order_by with order=desc."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||
dep = UserOrderCrud.order_params()
|
||||
result = await dep(order_by="username", order="desc")
|
||||
assert isinstance(result, UnaryExpression)
|
||||
assert "DESC" in str(result)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_invalid_order_by_raises_invalid_order_field_error(self):
|
||||
"""Raises InvalidOrderFieldError for an unknown order_by value."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||
dep = UserOrderCrud.order_params()
|
||||
with pytest.raises(InvalidOrderFieldError) as exc_info:
|
||||
await dep(order_by="nonexistent", order="asc")
|
||||
assert exc_info.value.field == "nonexistent"
|
||||
assert "username" in exc_info.value.valid_fields
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_multiple_fields_all_resolve(self):
|
||||
"""All configured fields resolve correctly via order_by."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username, User.email])
|
||||
dep = UserOrderCrud.order_params()
|
||||
result_username = await dep(order_by="username", order="asc")
|
||||
result_email = await dep(order_by="email", order="desc")
|
||||
assert isinstance(result_username, ColumnElement)
|
||||
assert isinstance(result_email, ColumnElement)
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_order_params_integrates_with_get_multi(
|
||||
self, db_session: AsyncSession
|
||||
):
|
||||
"""order_params output is accepted by get_multi(order_by=...)."""
|
||||
UserOrderCrud = CrudFactory(User, order_fields=[User.username])
|
||||
await UserCrud.create(
|
||||
db_session, UserCreate(username="charlie", email="c@test.com")
|
||||
)
|
||||
await UserCrud.create(
|
||||
db_session, UserCreate(username="alice", email="a@test.com")
|
||||
)
|
||||
|
||||
dep = UserOrderCrud.order_params()
|
||||
order_by = await dep(order_by="username", order="asc")
|
||||
results = await UserOrderCrud.get_multi(db_session, order_by=order_by)
|
||||
|
||||
assert results[0].username == "alice"
|
||||
assert results[1].username == "charlie"
|
||||
|
||||
@@ -15,12 +15,14 @@ from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_asyn
|
||||
from docs_src.examples.pagination_search.db import get_db
|
||||
from docs_src.examples.pagination_search.models import Article, Base, Category
|
||||
from docs_src.examples.pagination_search.routes import router
|
||||
from fastapi_toolsets.exceptions import init_exceptions_handlers
|
||||
|
||||
from .conftest import DATABASE_URL
|
||||
|
||||
|
||||
def build_app(session: AsyncSession) -> FastAPI:
|
||||
app = FastAPI()
|
||||
init_exceptions_handlers(app)
|
||||
|
||||
async def override_get_db():
|
||||
yield session
|
||||
@@ -269,3 +271,125 @@ class TestCursorPagination:
|
||||
body = resp.json()
|
||||
assert len(body["data"]) == 1
|
||||
assert body["data"][0]["title"] == "SQLAlchemy async"
|
||||
|
||||
|
||||
class TestOffsetSorting:
|
||||
"""Tests for order_by / order query parameters on the offset endpoint."""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_default_order_uses_created_at_asc(
|
||||
self, client: AsyncClient, ex_db_session
|
||||
):
|
||||
"""No order_by → default field (created_at) ASC."""
|
||||
await seed(ex_db_session)
|
||||
|
||||
resp = await client.get("/articles/offset")
|
||||
|
||||
assert resp.status_code == 200
|
||||
titles = [a["title"] for a in resp.json()["data"]]
|
||||
assert titles == ["FastAPI tips", "SQLAlchemy async", "Draft notes"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_order_by_title_asc(self, client: AsyncClient, ex_db_session):
|
||||
"""order_by=title&order=asc returns alphabetical order."""
|
||||
await seed(ex_db_session)
|
||||
|
||||
resp = await client.get("/articles/offset?order_by=title&order=asc")
|
||||
|
||||
assert resp.status_code == 200
|
||||
titles = [a["title"] for a in resp.json()["data"]]
|
||||
assert titles == ["Draft notes", "FastAPI tips", "SQLAlchemy async"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_order_by_title_desc(self, client: AsyncClient, ex_db_session):
|
||||
"""order_by=title&order=desc returns reverse alphabetical order."""
|
||||
await seed(ex_db_session)
|
||||
|
||||
resp = await client.get("/articles/offset?order_by=title&order=desc")
|
||||
|
||||
assert resp.status_code == 200
|
||||
titles = [a["title"] for a in resp.json()["data"]]
|
||||
assert titles == ["SQLAlchemy async", "FastAPI tips", "Draft notes"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_order_by_created_at_desc(self, client: AsyncClient, ex_db_session):
|
||||
"""order_by=created_at&order=desc returns newest-first."""
|
||||
await seed(ex_db_session)
|
||||
|
||||
resp = await client.get("/articles/offset?order_by=created_at&order=desc")
|
||||
|
||||
assert resp.status_code == 200
|
||||
titles = [a["title"] for a in resp.json()["data"]]
|
||||
assert titles == ["Draft notes", "SQLAlchemy async", "FastAPI tips"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_invalid_order_by_returns_422(
|
||||
self, client: AsyncClient, ex_db_session
|
||||
):
|
||||
"""Unknown order_by field returns 422 with SORT-422 error code."""
|
||||
resp = await client.get("/articles/offset?order_by=nonexistent_field")
|
||||
|
||||
assert resp.status_code == 422
|
||||
body = resp.json()
|
||||
assert body["error_code"] == "SORT-422"
|
||||
assert body["status"] == "FAIL"
|
||||
|
||||
|
||||
class TestCursorSorting:
|
||||
"""Tests for order_by / order query parameters on the cursor endpoint.
|
||||
|
||||
In cursor_paginate the cursor_column is always the primary sort; order_by
|
||||
acts as a secondary tiebreaker. With the seeded articles (all having unique
|
||||
created_at values) the overall ordering is always created_at ASC regardless
|
||||
of the order_by value — only the valid/invalid field check and the response
|
||||
shape are meaningful here.
|
||||
"""
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_default_order_uses_created_at_asc(
|
||||
self, client: AsyncClient, ex_db_session
|
||||
):
|
||||
"""No order_by → default field (created_at) ASC."""
|
||||
await seed(ex_db_session)
|
||||
|
||||
resp = await client.get("/articles/cursor")
|
||||
|
||||
assert resp.status_code == 200
|
||||
titles = [a["title"] for a in resp.json()["data"]]
|
||||
assert titles == ["FastAPI tips", "SQLAlchemy async", "Draft notes"]
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_order_by_title_asc_accepted(
|
||||
self, client: AsyncClient, ex_db_session
|
||||
):
|
||||
"""order_by=title is a valid field — request succeeds and returns all articles."""
|
||||
await seed(ex_db_session)
|
||||
|
||||
resp = await client.get("/articles/cursor?order_by=title&order=asc")
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()["data"]) == 3
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_order_by_title_desc_accepted(
|
||||
self, client: AsyncClient, ex_db_session
|
||||
):
|
||||
"""order_by=title&order=desc is valid — request succeeds and returns all articles."""
|
||||
await seed(ex_db_session)
|
||||
|
||||
resp = await client.get("/articles/cursor?order_by=title&order=desc")
|
||||
|
||||
assert resp.status_code == 200
|
||||
assert len(resp.json()["data"]) == 3
|
||||
|
||||
@pytest.mark.anyio
|
||||
async def test_invalid_order_by_returns_422(
|
||||
self, client: AsyncClient, ex_db_session
|
||||
):
|
||||
"""Unknown order_by field returns 422 with SORT-422 error code."""
|
||||
resp = await client.get("/articles/cursor?order_by=nonexistent_field")
|
||||
|
||||
assert resp.status_code == 422
|
||||
body = resp.json()
|
||||
assert body["error_code"] == "SORT-422"
|
||||
assert body["status"] == "FAIL"
|
||||
|
||||
@@ -8,6 +8,7 @@ from fastapi_toolsets.exceptions import (
|
||||
ApiException,
|
||||
ConflictError,
|
||||
ForbiddenError,
|
||||
InvalidOrderFieldError,
|
||||
NotFoundError,
|
||||
UnauthorizedError,
|
||||
generate_error_responses,
|
||||
@@ -334,3 +335,43 @@ class TestExceptionIntegration:
|
||||
|
||||
assert response.status_code == 200
|
||||
assert response.json() == {"id": 1}
|
||||
|
||||
|
||||
class TestInvalidOrderFieldError:
|
||||
"""Tests for InvalidOrderFieldError exception."""
|
||||
|
||||
def test_api_error_attributes(self):
|
||||
"""InvalidOrderFieldError has correct api_error metadata."""
|
||||
assert InvalidOrderFieldError.api_error.code == 422
|
||||
assert InvalidOrderFieldError.api_error.err_code == "SORT-422"
|
||||
assert InvalidOrderFieldError.api_error.msg == "Invalid Order Field"
|
||||
|
||||
def test_stores_field_and_valid_fields(self):
|
||||
"""InvalidOrderFieldError stores field and valid_fields on the instance."""
|
||||
error = InvalidOrderFieldError("unknown", ["name", "created_at"])
|
||||
assert error.field == "unknown"
|
||||
assert error.valid_fields == ["name", "created_at"]
|
||||
|
||||
def test_message_contains_field_and_valid_fields(self):
|
||||
"""Exception message mentions the bad field and valid options."""
|
||||
error = InvalidOrderFieldError("bad_field", ["name", "email"])
|
||||
assert "bad_field" in str(error)
|
||||
assert "name" in str(error)
|
||||
assert "email" in str(error)
|
||||
|
||||
def test_handled_as_422_by_exception_handler(self):
|
||||
"""init_exceptions_handlers turns InvalidOrderFieldError into a 422 response."""
|
||||
app = FastAPI()
|
||||
init_exceptions_handlers(app)
|
||||
|
||||
@app.get("/items")
|
||||
async def list_items():
|
||||
raise InvalidOrderFieldError("bad", ["name"])
|
||||
|
||||
client = TestClient(app)
|
||||
response = client.get("/items")
|
||||
|
||||
assert response.status_code == 422
|
||||
data = response.json()
|
||||
assert data["error_code"] == "SORT-422"
|
||||
assert data["status"] == "FAIL"
|
||||
|
||||
Reference in New Issue
Block a user