From 32059dcb02d0de499e70f0cc88efd49d850fbad7 Mon Sep 17 00:00:00 2001 From: d3vyce <44915747+d3vyce@users.noreply.github.com> Date: Wed, 1 Apr 2026 20:53:14 +0200 Subject: [PATCH] feat: consolidate *_params dependencies into per-paginate-style methods with feature toggles (#209) --- docs/examples/pagination-search.md | 16 +- docs/module/crud.md | 156 +---- docs_src/examples/pagination_search/routes.py | 52 +- src/fastapi_toolsets/crud/factory.py | 539 ++++++++++-------- tests/test_crud_search.py | 441 ++++++++------ uv.lock | 6 +- 6 files changed, 653 insertions(+), 557 deletions(-) diff --git a/docs/examples/pagination-search.md b/docs/examples/pagination-search.md index 28ee84b..1995673 100644 --- a/docs/examples/pagination-search.md +++ b/docs/examples/pagination-search.md @@ -43,16 +43,16 @@ Declare `searchable_fields`, `facet_fields`, and `order_fields` once on [`CrudFa ## Routes -```python title="routes.py:1:17" ---8<-- "docs_src/examples/pagination_search/routes.py:1:17" +```python title="routes.py:1:16" +--8<-- "docs_src/examples/pagination_search/routes.py:1:16" ``` ### Offset pagination Best for admin panels or any UI that needs a total item count and numbered pages. -```python title="routes.py:20:40" ---8<-- "docs_src/examples/pagination_search/routes.py:20:40" +```python title="routes.py:19:37" +--8<-- "docs_src/examples/pagination_search/routes.py:19:37" ``` **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. -```python title="routes.py:43:63" ---8<-- "docs_src/examples/pagination_search/routes.py:43:63" +```python title="routes.py:40:58" +--8<-- "docs_src/examples/pagination_search/routes.py:40:58" ``` **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. -```python title="routes.py:66:90" ---8<-- "docs_src/examples/pagination_search/routes.py:66:90" +```python title="routes.py:61:79" +--8<-- "docs_src/examples/pagination_search/routes.py:61:79" ``` **Offset request** (default) diff --git a/docs/module/crud.md b/docs/module/crud.md index ff711dc..fe12508 100644 --- a/docs/module/crud.md +++ b/docs/module/crud.md @@ -159,18 +159,15 @@ Three pagination methods are available. All return a typed response whose `pagi ### Offset pagination ```python +from typing import Annotated +from fastapi import Depends + @router.get("") async def get_users( session: SessionDep, - items_per_page: int = 50, - page: int = 1, + params: Annotated[dict, Depends(UserCrud.offset_paginate_params())], ) -> OffsetPaginatedResponse[UserRead]: - return await UserCrud.offset_paginate( - session=session, - items_per_page=items_per_page, - page=page, - schema=UserRead, - ) + return await UserCrud.offset_paginate(session=session, **params, 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): @@ -194,32 +191,13 @@ The [`offset_paginate`](../reference/crud.md#fastapi_toolsets.crud.factory.Async !!! 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 -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("") -async def list_users( +async def get_users( 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]: return await UserCrud.offset_paginate(session=session, **params, schema=UserRead) ``` @@ -230,15 +208,9 @@ async def list_users( @router.get("") async def list_users( session: SessionDep, - cursor: str | None = None, - items_per_page: int = 20, + params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())], ) -> CursorPaginatedResponse[UserRead]: - return await UserCrud.cursor_paginate( - session=session, - cursor=cursor, - items_per_page=items_per_page, - schema=UserRead, - ) + return await UserCrud.cursor_paginate(session=session, **params, 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): @@ -291,24 +263,6 @@ PostCrud = CrudFactory(model=Post, cursor_column=Post.id) 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) !!! 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. ```python -from fastapi_toolsets.crud import PaginationType from fastapi_toolsets.schemas import PaginatedResponse @router.get("") async def list_users( session: SessionDep, - pagination_type: PaginationType = PaginationType.OFFSET, - page: int = Query(1, ge=1, description="Current page (offset only)"), - cursor: str | None = Query(None, description="Cursor token (cursor only)"), - items_per_page: int = Query(20, ge=1, le=100), + params: Annotated[dict, Depends(UserCrud.paginate_params())], ) -> PaginatedResponse[UserRead]: - return await UserCrud.paginate( - session, - pagination_type=pagination_type, - page=page, - cursor=cursor, - items_per_page=items_per_page, - schema=UserRead, - ) + return await UserCrud.paginate(session, **params, 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 ``` -#### 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 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("") async def get_users( session: SessionDep, - items_per_page: int = 50, - page: int = 1, - search: str | None = None, + params: Annotated[dict, Depends(UserCrud.offset_paginate_params())], ) -> OffsetPaginatedResponse[UserRead]: - return await UserCrud.offset_paginate( - session=session, - items_per_page=items_per_page, - page=page, - search=search, - schema=UserRead, - ) + return await UserCrud.offset_paginate(session=session, **params, schema=UserRead) ``` ```python @router.get("") async def get_users( session: SessionDep, - cursor: str | None = None, - items_per_page: int = 50, - search: str | None = None, + params: Annotated[dict, Depends(UserCrud.cursor_paginate_params())], ) -> CursorPaginatedResponse[UserRead]: - return await UserCrud.cursor_paginate( - session=session, - items_per_page=items_per_page, - cursor=cursor, - search=search, - schema=UserRead, - ) + return await UserCrud.cursor_paginate(session=session, **params, schema=UserRead) ``` ### 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. -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 from typing import Annotated @@ -501,13 +409,11 @@ UserCrud = CrudFactory( @router.get("", response_model_exclude_none=True) async def list_users( session: SessionDep, - page: int = 1, - filter_by: Annotated[dict[str, list[str]], Depends(UserCrud.filter_params())], + params: Annotated[dict, Depends(UserCrud.offset_paginate_params())], ) -> OffsetPaginatedResponse[UserRead]: return await UserCrud.offset_paginate( session=session, - page=page, - filter_by=filter_by, + **params, 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 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())], + params: Annotated[dict, Depends(UserCrud.offset_paginate_params( + default_order_field=User.created_at, + ))], ) -> 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: @@ -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). -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 -UserOrderParams = UserCrud.order_params(order_fields=[User.name]) +params = UserCrud.offset_paginate_params(order_fields=[User.name]) ``` ## Relationship loading @@ -656,12 +563,11 @@ async def get_user(session: SessionDep, uuid: UUID) -> Response[UserRead]: ) @router.get("") -async def list_users(session: SessionDep, page: int = 1) -> OffsetPaginatedResponse[UserRead]: - return await crud.UserCrud.offset_paginate( - session=session, - page=page, - schema=UserRead, - ) +async def list_users( + session: SessionDep, + params: Annotated[dict, Depends(crud.UserCrud.offset_paginate_params())], +) -> OffsetPaginatedResponse[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. diff --git a/docs_src/examples/pagination_search/routes.py b/docs_src/examples/pagination_search/routes.py index 2c00f3b..a35da4f 100644 --- a/docs_src/examples/pagination_search/routes.py +++ b/docs_src/examples/pagination_search/routes.py @@ -2,7 +2,6 @@ from typing import Annotated from fastapi import APIRouter, Depends -from fastapi_toolsets.crud import OrderByClause from fastapi_toolsets.schemas import ( CursorPaginatedResponse, OffsetPaginatedResponse, @@ -22,21 +21,18 @@ async def list_articles_offset( session: SessionDep, params: Annotated[ 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]: return await ArticleCrud.offset_paginate( session=session, **params, - search=search, - filter_by=filter_by or None, - order_by=order_by, schema=ArticleRead, ) @@ -46,21 +42,18 @@ async def list_articles_cursor( session: SessionDep, params: Annotated[ 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]: return await ArticleCrud.cursor_paginate( session=session, **params, - search=search, - filter_by=filter_by or None, - order_by=order_by, schema=ArticleRead, ) @@ -70,20 +63,17 @@ async def list_articles( session: SessionDep, params: Annotated[ 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]: return await ArticleCrud.paginate( session, **params, - search=search, - filter_by=filter_by or None, - order_by=order_by, schema=ArticleRead, ) diff --git a/src/fastapi_toolsets/crud/factory.py b/src/fastapi_toolsets/crud/factory.py index 5fde829..1707e3b 100644 --- a/src/fastapi_toolsets/crud/factory.py +++ b/src/fastapi_toolsets/crud/factory.py @@ -263,105 +263,6 @@ class AsyncCrud(Generic[ModelType]): base_joins=search_joins, ) - @classmethod - def filter_params( - cls: type[Self], - *, - facet_fields: Sequence[FacetFieldType] | None = None, - ) -> Callable[..., Awaitable[dict[str, list[str]]]]: - """Return a FastAPI dependency that collects facet filter values from query parameters. - - Args: - facet_fields: Override the facet fields for this dependency. Falls back to the - class-level ``facet_fields`` if not provided. - - Returns: - An async dependency function named ``{Model}FilterParams`` that resolves to a - ``dict[str, list[str]]`` containing only the keys that were supplied in the - request (absent/``None`` parameters are excluded). - - Raises: - ValueError: If no facet fields are configured on this CRUD class and none are - provided via ``facet_fields``. - """ - fields = cls._resolve_facet_fields(facet_fields) - if not fields: - raise ValueError( - f"{cls.__name__} has no facet_fields configured. " - "Pass facet_fields= or set them on CrudFactory." - ) - keys = facet_keys(fields) - - async def dependency(**kwargs: Any) -> dict[str, list[str]]: - return {k: v for k, v in kwargs.items() if v is not None} - - dependency.__name__ = f"{cls.model.__name__}FilterParams" - dependency.__signature__ = inspect.Signature( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute] - parameters=[ - inspect.Parameter( - k, - inspect.Parameter.KEYWORD_ONLY, - annotation=list[str] | None, - default=Query(default=None), - ) - for k in keys - ] - ) - - return dependency - - @classmethod - def search_params( - cls: type[Self], - *, - search_fields: Sequence[SearchFieldType] | None = None, - ) -> Callable[..., Awaitable[dict[str, Any]]]: - """Return a FastAPI dependency that collects search params from query parameters. - - Args: - search_fields: Override search fields for this dependency. - Falls back to the class-level ``searchable_fields``. - - Returns: - An async dependency function named ``{Model}SearchParams`` that - resolves to a ``dict`` with ``search`` and ``search_column`` keys - (absent keys are excluded). - """ - fields = search_fields if search_fields is not None else cls.searchable_fields - if not fields: - raise ValueError( - f"{cls.__name__} has no searchable_fields configured. " - "Pass search_fields= or set them on CrudFactory." - ) - keys = search_field_keys(fields) - - async def dependency(**kwargs: Any) -> dict[str, Any]: - return {k: v for k, v in kwargs.items() if v is not None} - - dependency.__name__ = f"{cls.model.__name__}SearchParams" - dependency.__signature__ = inspect.Signature( # type: ignore[attr-defined] # ty:ignore[unresolved-attribute] - parameters=[ - inspect.Parameter( - "search", - inspect.Parameter.KEYWORD_ONLY, - annotation=str | None, - default=Query(default=None, description="Search query string"), - ), - inspect.Parameter( - "search_column", - inspect.Parameter.KEYWORD_ONLY, - annotation=str | None, - default=Query( - default=None, - description="Restrict search to a single column", - enum=keys, - ), - ), - ] - ) - - return dependency - @classmethod def _resolve_search_columns( cls: type[Self], @@ -374,71 +275,274 @@ class AsyncCrud(Generic[ModelType]): return search_field_keys(fields) @classmethod - def offset_params( + def _build_paginate_params( + cls: type[Self], + *, + pagination_params: list[inspect.Parameter], + pagination_fixed: dict[str, Any], + 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) + + search_keys: list[str] | None = None + if search: + search_keys = cls._resolve_search_columns(search_fields) + if search_keys: + all_params.extend( + [ + inspect.Parameter( + "search", + inspect.Parameter.KEYWORD_ONLY, + annotation=str | None, + default=Query( + default=None, description="Search query string" + ), + ), + inspect.Parameter( + "search_column", + inspect.Parameter.KEYWORD_ONLY, + annotation=str | None, + default=Query( + default=None, + description="Restrict search to a single column", + enum=search_keys, + ), + ), + ] + ) + reserved_names.update({"search", "search_column"}) + + filter_keys: list[str] | None = None + if filter: + resolved_facets = cls._resolve_facet_fields(facet_fields) + if resolved_facets: + filter_keys = facet_keys(resolved_facets) + 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( + k, + inspect.Parameter.KEYWORD_ONLY, + annotation=list[str] | None, + default=Query(default=None), + ) + 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 + + @classmethod + def offset_paginate_params( cls: type[Self], *, default_page_size: int = 20, max_page_size: int = 100, 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]]]: - """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: - default_page_size: Default value for the ``items_per_page`` query parameter. - max_page_size: Maximum allowed value for ``items_per_page`` (enforced via - ``le`` on the ``Query``). - include_total: Server-side flag forwarded as-is to ``include_total`` in - :meth:`offset_paginate`. Not exposed as a query parameter. + default_page_size: Default ``items_per_page`` value. + max_page_size: Maximum ``items_per_page`` value. + include_total: Whether to include total count (not a query param). + 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: - An async dependency that resolves to a dict with ``page``, - ``items_per_page``, and ``include_total`` keys, ready to be - unpacked into :meth:`offset_paginate`. + An async dependency that resolves to a dict ready to be unpacked + into :meth:`offset_paginate`. """ - - async def dependency( - page: int = Query(1, ge=1, description="Page number (1-indexed)"), - items_per_page: int = _page_size_query(default_page_size, max_page_size), - ) -> dict[str, Any]: - return { - "page": page, - "items_per_page": items_per_page, - "include_total": include_total, - } - - dependency.__name__ = f"{cls.model.__name__}OffsetParams" - return dependency + pagination_params = [ + inspect.Parameter( + "page", + inspect.Parameter.KEYWORD_ONLY, + annotation=int, + default=Query(1, ge=1, description="Page number (1-indexed)"), + ), + inspect.Parameter( + "items_per_page", + inspect.Parameter.KEYWORD_ONLY, + annotation=int, + default=_page_size_query(default_page_size, max_page_size), + ), + ] + 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 - def cursor_params( + def cursor_paginate_params( cls: type[Self], *, default_page_size: int = 20, 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]]]: - """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: - default_page_size: Default value for the ``items_per_page`` query parameter. - max_page_size: Maximum allowed value for ``items_per_page`` (enforced via - ``le`` on the ``Query``). + default_page_size: Default ``items_per_page`` value. + max_page_size: Maximum ``items_per_page`` value. + 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: - An async dependency that resolves to a dict with ``cursor`` and - ``items_per_page`` keys, ready to be unpacked into - :meth:`cursor_paginate`. + An async dependency that resolves to a dict ready to be unpacked + into :meth:`cursor_paginate`. """ - - async def dependency( - cursor: str | None = Query( - None, description="Cursor token from a previous response" + pagination_params = [ + inspect.Parameter( + "cursor", + inspect.Parameter.KEYWORD_ONLY, + annotation=str | None, + default=Query( + None, description="Cursor token from a previous response" + ), ), - items_per_page: int = _page_size_query(default_page_size, max_page_size), - ) -> dict[str, Any]: - return {"cursor": cursor, "items_per_page": items_per_page} - - dependency.__name__ = f"{cls.model.__name__}CursorParams" - return dependency + inspect.Parameter( + "items_per_page", + inspect.Parameter.KEYWORD_ONLY, + annotation=int, + 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 def paginate_params( @@ -448,102 +552,81 @@ class AsyncCrud(Generic[ModelType]): max_page_size: int = 100, default_pagination_type: PaginationType = PaginationType.OFFSET, include_total: bool = True, - ) -> Callable[..., Awaitable[dict[str, Any]]]: - """Return a FastAPI dependency that collects all pagination params from query params. - - Args: - default_page_size: Default value for the ``items_per_page`` query parameter. - max_page_size: Maximum allowed value for ``items_per_page`` (enforced via - ``le`` on the ``Query``). - default_pagination_type: Default pagination strategy. - include_total: Server-side flag forwarded as-is to ``include_total`` in - :meth:`paginate`. Not exposed as a query parameter. - - Returns: - An async dependency that resolves to a dict with ``pagination_type``, - ``page``, ``cursor``, ``items_per_page``, and ``include_total`` keys, - ready to be unpacked into :meth:`paginate`. - """ - - async def dependency( - pagination_type: PaginationType = Query( - default_pagination_type, description="Pagination strategy" - ), - page: int = Query( - 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), - ) -> dict[str, Any]: - return { - "pagination_type": pagination_type, - "page": page, - "cursor": cursor, - "items_per_page": items_per_page, - "include_total": include_total, - } - - dependency.__name__ = f"{cls.model.__name__}PaginateParams" - return dependency - - @classmethod - def order_params( - cls: type[Self], - *, + 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_field: QueryableAttribute[Any] | None = None, + default_order_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. + ) -> Callable[..., Awaitable[dict[str, Any]]]: + """Return a FastAPI dependency that collects all params for :meth:`paginate`. 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"``). + default_page_size: Default ``items_per_page`` value. + max_page_size: Maximum ``items_per_page`` value. + default_pagination_type: Default pagination strategy. + include_total: Whether to include total count (not a query param). + 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: - 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. + An async dependency that resolves to a dict ready to be unpacked + into :meth:`paginate`. """ - 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}" + pagination_params = [ + inspect.Parameter( + "pagination_type", + inspect.Parameter.KEYWORD_ONLY, + annotation=PaginationType, + default=Query( + default_pagination_type, description="Pagination strategy" + ), ), - order: Literal["asc", "desc"] = Query( - default_order, description="Sort direction" + inspect.Parameter( + "page", + inspect.Parameter.KEYWORD_ONLY, + annotation=int, + default=Query( + 1, ge=1, description="Page number (1-indexed, offset only)" + ), ), - ) -> 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 + inspect.Parameter( + "cursor", + inspect.Parameter.KEYWORD_ONLY, + annotation=str | None, + default=Query( + None, + description="Cursor token from a previous response (cursor only)", + ), + ), + inspect.Parameter( + "items_per_page", + inspect.Parameter.KEYWORD_ONLY, + annotation=int, + default=_page_size_query(default_page_size, max_page_size), + ), + ] + return cls._build_paginate_params( + pagination_params=pagination_params, + pagination_fixed={"include_total": include_total}, + dep_name=f"{cls.model.__name__}PaginateParams", + 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, + ) @overload @classmethod @@ -1568,7 +1651,7 @@ def CrudFactory( 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. + via ``offset_paginate_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. diff --git a/tests/test_crud_search.py b/tests/test_crud_search.py index 9e88171..3d6591a 100644 --- a/tests/test_crud_search.py +++ b/tests/test_crud_search.py @@ -1031,61 +1031,64 @@ class TestFilterBy: assert "JSON" in exc_info.value.col_type -class TestFilterParamsSchema: - """Tests for AsyncCrud.filter_params().""" +class TestFilterParamsViaConsolidated: + """Tests for filter params via consolidated offset_paginate_params().""" def test_generates_fields_from_facet_fields(self): """Returned dependency has one keyword param per facet field.""" - import inspect - 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) - assert param_names == {"username", "email"} + assert "username" in param_names + assert "email" in param_names def test_relationship_facet_uses_full_chain_key(self): """Relationship tuple uses the full chain joined by __ as the key.""" - import inspect - 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) - assert param_names == {"role__name"} + assert "role__name" in param_names - def test_raises_when_no_facet_fields(self): - """ValueError raised when no facet_fields are configured or provided.""" - with pytest.raises(ValueError, match="no facet_fields"): - UserCrud.filter_params() + def test_filter_disabled_no_facet_params(self): + """When filter=False, no facet params are generated.""" + UserFacetCrud = CrudFactory(User, facet_fields=[User.username, User.email]) + 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): """facet_fields= parameter overrides the class-level default.""" - import inspect - 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) - assert param_names == {"email"} + assert "email" in param_names + assert "username" not in param_names @pytest.mark.anyio - async def test_awaiting_dep_returns_dict_with_values(self): - """Awaiting the dependency returns a dict with only the supplied keys.""" + async def test_awaiting_dep_returns_filter_by_with_values(self): + """Awaiting the dependency returns filter_by dict with only supplied keys.""" 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"]) - assert result == {"username": ["alice"]} + result = await dep(page=1, items_per_page=20, username=["alice"], email=None) + assert result["filter_by"] == {"username": ["alice"]} @pytest.mark.anyio async def test_multi_value_list_field(self): """Multiple values are accepted as a list.""" 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"]) - assert result == {"username": ["alice", "bob"]} + result = await dep(page=1, items_per_page=20, username=["alice", "bob"]) + assert result["filter_by"] == {"username": ["alice", "bob"]} def test_disambiguates_duplicate_column_keys(self): """Two relationship tuples sharing a terminal column key get prefixed names.""" @@ -1130,15 +1133,15 @@ class TestFilterParamsSchema: assert keys == ["username", "email"] 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]) - 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 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]) await UserCrud.create( db_session, UserCreate(username="alice", email="a@test.com") @@ -1147,10 +1150,10 @@ class TestFilterParamsSchema: db_session, UserCreate(username="bob", email="b@test.com") ) - dep = UserFacetCrud.filter_params() - f = await dep(username=["alice"]) + dep = UserFacetCrud.offset_paginate_params(search=False, order=False) + params = await dep(page=1, items_per_page=20, username=["alice"]) result = await UserFacetCrud.offset_paginate( - db_session, filter_by=f, schema=UserRead + db_session, **params, schema=UserRead ) assert isinstance(result.pagination, OffsetPagination) @@ -1159,7 +1162,7 @@ class TestFilterParamsSchema: @pytest.mark.anyio 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( User, cursor_column=User.id, facet_fields=[User.username] ) @@ -1170,10 +1173,10 @@ class TestFilterParamsSchema: db_session, UserCreate(username="bob", email="b@test.com") ) - dep = UserFacetCursorCrud.filter_params() - f = await dep(username=["alice"]) + dep = UserFacetCursorCrud.cursor_paginate_params(search=False, order=False) + params = await dep(cursor=None, items_per_page=20, username=["alice"]) result = await UserFacetCursorCrud.cursor_paginate( - db_session, filter_by=f, schema=UserRead + db_session, **params, schema=UserRead ) assert len(result.data) == 1 @@ -1181,7 +1184,7 @@ class TestFilterParamsSchema: @pytest.mark.anyio 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]) await UserCrud.create( db_session, UserCreate(username="alice", email="a@test.com") @@ -1190,53 +1193,70 @@ class TestFilterParamsSchema: db_session, UserCreate(username="bob", email="b@test.com") ) - dep = UserFacetCrud.filter_params() - f = await dep() # all fields None + dep = UserFacetCrud.offset_paginate_params(search=False, order=False) + params = await dep(page=1, items_per_page=20) # all facet fields None + assert params["filter_by"] is None result = await UserFacetCrud.offset_paginate( - db_session, filter_by=f, schema=UserRead + db_session, **params, schema=UserRead ) assert isinstance(result.pagination, OffsetPagination) 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 TestSearchParamsSchema: - """Tests for AsyncCrud.search_params().""" + mock_field = MagicMock() + 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.search_params() + dep = UserSearchCrud.offset_paginate_params(filter=False, order=False) param_names = set(inspect.signature(dep).parameters) - assert param_names == {"search", "search_column"} + assert "search" in param_names + assert "search_column" in param_names - def test_dependency_name_includes_model_name(self): - """Dependency function is named {Model}SearchParams.""" + 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.search_params() - assert dep.__name__ == "UserSearchParams" # type: ignore[union-attr] # ty:ignore[unresolved-attribute] + dep = UserSearchCrud.offset_paginate_params( + search=False, filter=False, order=False + ) - def test_raises_when_no_searchable_fields(self): - """ValueError raised when overriding with empty search_fields.""" - UserSearchCrud = CrudFactory(User, searchable_fields=[User.username]) - with pytest.raises(ValueError, match="no searchable_fields"): - UserSearchCrud.search_params(search_fields=[]) + 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 a dict with search key.""" + """Awaiting the dependency with only search returns search in dict.""" UserSearchCrud = CrudFactory( User, searchable_fields=[User.username, User.email] ) - dep = UserSearchCrud.search_params() + dep = UserSearchCrud.offset_paginate_params(filter=False, order=False) - result = await dep(search="alice") - assert result == {"search": "alice"} + 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): @@ -1244,28 +1264,32 @@ class TestSearchParamsSchema: UserSearchCrud = CrudFactory( User, searchable_fields=[User.username, User.email] ) - dep = UserSearchCrud.search_params() + dep = UserSearchCrud.offset_paginate_params(filter=False, order=False) - result = await dep(search="alice", search_column="username") - assert result == {"search": "alice", "search_column": "username"} + 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_values(self): - """Awaiting the dependency with no values returns an empty dict.""" + 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.search_params() + dep = UserSearchCrud.offset_paginate_params(filter=False, order=False) - result = await dep() - assert result == {} + 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.search_params() + dep = UserRelSearchCrud.offset_paginate_params(filter=False, order=False) params = inspect.signature(dep).parameters search_column_param = params["search_column"] @@ -1402,36 +1426,36 @@ class TestSearchColumns: assert result.data[0].username == "bob" -class TestOrderParamsSchema: - """Tests for AsyncCrud.order_params().""" +class TestOrderParamsViaConsolidated: + """Tests for order params via consolidated offset_paginate_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() + dep = UserOrderCrud.offset_paginate_params(search=False, filter=False) 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): - """Dependency function is named after the model.""" + def test_order_disabled_no_order_params(self): + """When order=False, no order params are generated.""" UserOrderCrud = CrudFactory(User, order_fields=[User.username]) - dep = UserOrderCrud.order_params() - assert getattr(dep, "__name__") == "UserOrderParams" + dep = UserOrderCrud.offset_paginate_params( + search=False, filter=False, order=False + ) - 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() + param_names = set(inspect.signature(dep).parameters) + assert "order_by" not in param_names + assert "order" not in param_names 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]) + 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) description = sig.parameters["order_by"].default.description assert "email" in description @@ -1440,7 +1464,7 @@ class TestOrderParamsSchema: 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() + dep = UserOrderCrud.offset_paginate_params(search=False, filter=False) sig = inspect.signature(dep) description = sig.parameters["order_by"].default.description @@ -1450,8 +1474,12 @@ class TestOrderParamsSchema: 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") + dep_asc = UserOrderCrud.offset_paginate_params( + 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_desc = inspect.signature(dep_desc) @@ -1460,55 +1488,59 @@ class TestOrderParamsSchema: @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.""" + """Returns order_by=None when order_by is absent and no default_order_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 + dep = UserOrderCrud.offset_paginate_params(search=False, filter=False) + result = await dep(page=1, items_per_page=20, order_by=None, order="asc") + assert result["order_by"] 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.""" + """Returns default_order_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) + dep = UserOrderCrud.offset_paginate_params( + search=False, filter=False, default_order_field=User.username + ) + 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 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]) - 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) + dep = UserOrderCrud.offset_paginate_params( + search=False, filter=False, default_order_field=User.username + ) + 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 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) + dep = UserOrderCrud.offset_paginate_params(search=False, filter=False) + result = await dep(page=1, items_per_page=20, order_by="username", order="asc") + assert isinstance(result["order_by"], UnaryExpression) + assert "ASC" in str(result["order_by"]) @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) + dep = UserOrderCrud.offset_paginate_params(search=False, filter=False) + result = await dep(page=1, items_per_page=20, order_by="username", order="desc") + assert isinstance(result["order_by"], UnaryExpression) + assert "DESC" in str(result["order_by"]) @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() + dep = UserOrderCrud.offset_paginate_params(search=False, filter=False) 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 "username" in exc_info.value.valid_fields @@ -1516,17 +1548,21 @@ class TestOrderParamsSchema: 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) + dep = UserOrderCrud.offset_paginate_params(search=False, filter=False) + result_username = await dep( + page=1, items_per_page=20, order_by="username", order="asc" + ) + 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 - async def test_order_params_integrates_with_get_multi( + async def test_order_integrates_with_offset_paginate( 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]) await UserCrud.create( db_session, UserCreate(username="charlie", email="c@test.com") @@ -1535,37 +1571,43 @@ class TestOrderParamsSchema: 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) + dep = UserOrderCrud.offset_paginate_params(search=False, filter=False) + params = await dep(page=1, items_per_page=20, order_by="username", order="asc") + result = await UserOrderCrud.offset_paginate( + db_session, **params, schema=UserRead + ) - assert results[0].username == "alice" - assert results[1].username == "charlie" + assert result.data[0].username == "alice" + assert result.data[1].username == "charlie" -class TestOffsetParamsSchema: - """Tests for AsyncCrud.offset_params().""" +class TestOffsetPaginateParamsSchema: + """Tests for AsyncCrud.offset_paginate_params().""" def test_returns_page_and_items_per_page_params(self): - """Returned dependency has page and items_per_page params only.""" - dep = RoleCrud.offset_params() + """Returned dependency has page and items_per_page params.""" + dep = RoleCrud.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_dependency_name_includes_model_name(self): """Dependency function is named after the model.""" - dep = RoleCrud.offset_params() - assert getattr(dep, "__name__") == "RoleOffsetParams" + dep = RoleCrud.offset_paginate_params(search=False, filter=False, order=False) + assert getattr(dep, "__name__") == "RoleOffsetPaginateParams" def test_default_page_size_reflected_in_items_per_page_default(self): """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) assert sig.parameters["items_per_page"].default.default == 42 def test_max_page_size_reflected_in_items_per_page_le(self): """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) le = next( m.le @@ -1576,67 +1618,121 @@ class TestOffsetParamsSchema: def test_include_total_not_a_query_param(self): """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) assert "include_total" not in param_names @pytest.mark.anyio async def test_include_total_true_forwarded_in_result(self): """include_total=True factory arg appears in the resolved dict.""" - result = await RoleCrud.offset_params(include_total=True)( - page=1, items_per_page=10 - ) + result = await RoleCrud.offset_paginate_params( + include_total=True, search=False, filter=False, order=False + )(page=1, items_per_page=10) assert result["include_total"] is True @pytest.mark.anyio async def test_include_total_false_forwarded_in_result(self): """include_total=False factory arg appears in the resolved dict.""" - result = await RoleCrud.offset_params(include_total=False)( - page=1, items_per_page=10 - ) + result = await RoleCrud.offset_paginate_params( + include_total=False, search=False, filter=False, order=False + )(page=1, items_per_page=10) assert result["include_total"] is False @pytest.mark.anyio async def test_awaiting_dep_returns_dict(self): """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) assert result == {"page": 2, "items_per_page": 10, "include_total": False} @pytest.mark.anyio 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")) - dep = RoleCrud.offset_params() + dep = RoleCrud.offset_paginate_params(search=False, filter=False, order=False) params = await dep(page=1, items_per_page=10) result = await RoleCrud.offset_paginate(db_session, **params, schema=RoleRead) assert result.pagination.page == 1 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: - """Tests for AsyncCrud.cursor_params().""" + def test_search_enabled_but_no_searchable_fields(self): + """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): """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) assert param_names == {"cursor", "items_per_page"} def test_dependency_name_includes_model_name(self): """Dependency function is named after the model.""" - dep = RoleCursorCrud.cursor_params() - assert getattr(dep, "__name__") == "RoleCursorParams" + dep = RoleCursorCrud.cursor_paginate_params( + search=False, filter=False, order=False + ) + assert getattr(dep, "__name__") == "RoleCursorPaginateParams" def test_default_page_size_reflected_in_items_per_page_default(self): """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) assert sig.parameters["items_per_page"].default.default == 15 def test_max_page_size_reflected_in_items_per_page_le(self): """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) le = next( m.le @@ -1647,22 +1743,28 @@ class TestCursorParamsSchema: def test_cursor_defaults_to_none(self): """cursor defaults to None.""" - dep = RoleCursorCrud.cursor_params() + dep = RoleCursorCrud.cursor_paginate_params( + search=False, filter=False, order=False + ) sig = inspect.signature(dep) assert sig.parameters["cursor"].default.default is None @pytest.mark.anyio async def test_awaiting_dep_returns_dict(self): """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) assert result == {"cursor": None, "items_per_page": 5} @pytest.mark.anyio 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")) - dep = RoleCursorCrud.cursor_params() + dep = RoleCursorCrud.cursor_paginate_params( + search=False, filter=False, order=False + ) params = await dep(cursor=None, items_per_page=10) result = await RoleCursorCrud.cursor_paginate( db_session, **params, schema=RoleRead @@ -1675,13 +1777,13 @@ class TestPaginateParamsSchema: def test_returns_all_params(self): """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) assert param_names == {"pagination_type", "page", "cursor", "items_per_page"} def test_dependency_name_includes_model_name(self): """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" def test_default_pagination_type(self): @@ -1689,7 +1791,10 @@ class TestPaginateParamsSchema: from fastapi_toolsets.schemas import PaginationType dep = RoleCursorCrud.paginate_params( - default_pagination_type=PaginationType.CURSOR + default_pagination_type=PaginationType.CURSOR, + search=False, + filter=False, + order=False, ) sig = inspect.signature(dep) assert ( @@ -1698,13 +1803,17 @@ class TestPaginateParamsSchema: def test_default_page_size(self): """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) assert sig.parameters["items_per_page"].default.default == 15 def test_max_page_size_le_constraint(self): """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) le = next( m.le @@ -1715,19 +1824,23 @@ class TestPaginateParamsSchema: def test_include_total_not_a_query_param(self): """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) @pytest.mark.anyio async def test_include_total_forwarded_in_result(self): """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, page=1, cursor=None, 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, page=1, cursor=None, @@ -1739,7 +1852,7 @@ class TestPaginateParamsSchema: @pytest.mark.anyio async def test_awaiting_dep_returns_dict(self): """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( pagination_type=PaginationType.OFFSET, page=2, @@ -1760,7 +1873,9 @@ class TestPaginateParamsSchema: from fastapi_toolsets.schemas import OffsetPagination 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, page=1, cursor=None, @@ -1775,7 +1890,9 @@ class TestPaginateParamsSchema: from fastapi_toolsets.schemas import CursorPagination 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, page=1, cursor=None, diff --git a/uv.lock b/uv.lock index 58d6ee1..8904dad 100644 --- a/uv.lock +++ b/uv.lock @@ -894,15 +894,15 @@ wheels = [ [[package]] name = "pymdown-extensions" -version = "10.21" +version = "10.21.2" source = { registry = "https://pypi.org/simple" } dependencies = [ { name = "markdown" }, { 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 = [ - { 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]]