diff --git a/docs/module/crud.md b/docs/module/crud.md index 15bc7c8..422d935 100644 --- a/docs/module/crud.md +++ b/docs/module/crud.md @@ -206,6 +206,24 @@ result = await UserCrud.offset_paginate( ) ``` +#### 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( + session: SessionDep, + params: Annotated[dict, Depends(UserCrud.offset_params(default_page_size=20, max_page_size=100))], +) -> OffsetPaginatedResponse[UserRead]: + return await UserCrud.offset_paginate(session=session, **params, schema=UserRead) +``` + ### Cursor pagination ```python @@ -273,6 +291,24 @@ 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`" @@ -306,7 +342,24 @@ GET /users?pagination_type=offset&page=2&items_per_page=10 GET /users?pagination_type=cursor&cursor=eyJ2YWx1ZSI6...&items_per_page=10 ``` -Both `page` and `cursor` are always accepted by the endpoint — unused parameters are silently ignored by `paginate()`. +#### 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 diff --git a/docs_src/examples/pagination_search/routes.py b/docs_src/examples/pagination_search/routes.py index bdc6857..2c00f3b 100644 --- a/docs_src/examples/pagination_search/routes.py +++ b/docs_src/examples/pagination_search/routes.py @@ -1,8 +1,8 @@ from typing import Annotated -from fastapi import APIRouter, Depends, Query +from fastapi import APIRouter, Depends -from fastapi_toolsets.crud import OrderByClause, PaginationType +from fastapi_toolsets.crud import OrderByClause from fastapi_toolsets.schemas import ( CursorPaginatedResponse, OffsetPaginatedResponse, @@ -20,19 +20,20 @@ router = APIRouter(prefix="/articles") @router.get("/offset") async def list_articles_offset( session: SessionDep, + params: Annotated[ + dict, + Depends(ArticleCrud.offset_params(default_page_size=20, max_page_size=100)), + ], 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, ) -> OffsetPaginatedResponse[ArticleRead]: return await ArticleCrud.offset_paginate( session=session, - page=page, - items_per_page=items_per_page, + **params, search=search, filter_by=filter_by or None, order_by=order_by, @@ -43,19 +44,20 @@ async def list_articles_offset( @router.get("/cursor") async def list_articles_cursor( session: SessionDep, + params: Annotated[ + dict, + Depends(ArticleCrud.cursor_params(default_page_size=20, max_page_size=100)), + ], 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, ) -> CursorPaginatedResponse[ArticleRead]: return await ArticleCrud.cursor_paginate( session=session, - cursor=cursor, - items_per_page=items_per_page, + **params, search=search, filter_by=filter_by or None, order_by=order_by, @@ -66,23 +68,20 @@ async def list_articles_cursor( @router.get("/") async def list_articles( session: SessionDep, + params: Annotated[ + dict, + Depends(ArticleCrud.paginate_params(default_page_size=20, max_page_size=100)), + ], filter_by: Annotated[dict[str, list[str]], Depends(ArticleCrud.filter_params())], order_by: Annotated[ OrderByClause | None, Depends(ArticleCrud.order_params(default_field=Article.created_at)), ], - pagination_type: PaginationType = PaginationType.OFFSET, - page: int = Query(1, ge=1), - cursor: str | None = None, - items_per_page: int = Query(20, ge=1, le=100), search: str | None = None, ) -> PaginatedResponse[ArticleRead]: return await ArticleCrud.paginate( session, - pagination_type=pagination_type, - page=page, - cursor=cursor, - items_per_page=items_per_page, + **params, search=search, filter_by=filter_by or None, order_by=order_by, diff --git a/src/fastapi_toolsets/crud/factory.py b/src/fastapi_toolsets/crud/factory.py index fdeced0..a041fed 100644 --- a/src/fastapi_toolsets/crud/factory.py +++ b/src/fastapi_toolsets/crud/factory.py @@ -75,6 +75,16 @@ def _decode_cursor(cursor: str) -> tuple[str, _CursorDirection]: return payload["val"], _CursorDirection(payload["dir"]) +def _page_size_query(default: int, max_size: int) -> int: + """Return a FastAPI ``Query`` for the ``items_per_page`` parameter.""" + return Query( + default, + ge=1, + le=max_size, + description=f"Number of items per page (max {max_size})", + ) + + def _parse_cursor_value(raw_val: str, col_type: Any) -> Any: """Parse a raw cursor string value back into the appropriate Python type.""" if isinstance(col_type, Integer): @@ -259,6 +269,7 @@ class AsyncCrud(Generic[ModelType]): 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. @@ -298,6 +309,121 @@ class AsyncCrud(Generic[ModelType]): return dependency + @classmethod + def offset_params( + cls: type[Self], + *, + default_page_size: int = 20, + max_page_size: int = 100, + include_total: bool = True, + ) -> Callable[..., Awaitable[dict[str, Any]]]: + """Return a FastAPI dependency that collects offset 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``). + include_total: Server-side flag forwarded as-is to ``include_total`` in + :meth:`offset_paginate`. Not exposed as a query parameter. + + 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`. + """ + + 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 + + @classmethod + def cursor_params( + cls: type[Self], + *, + default_page_size: int = 20, + max_page_size: int = 100, + ) -> Callable[..., Awaitable[dict[str, Any]]]: + """Return a FastAPI dependency that collects cursor 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``). + + Returns: + An async dependency that resolves to a dict with ``cursor`` and + ``items_per_page`` keys, ready to be unpacked into + :meth:`cursor_paginate`. + """ + + async def dependency( + cursor: str | None = 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 + + @classmethod + def paginate_params( + cls: type[Self], + *, + default_page_size: int = 20, + 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], diff --git a/tests/test_crud_search.py b/tests/test_crud_search.py index 9421379..84bde05 100644 --- a/tests/test_crud_search.py +++ b/tests/test_crud_search.py @@ -14,12 +14,14 @@ from fastapi_toolsets.crud import ( get_searchable_fields, ) from fastapi_toolsets.exceptions import InvalidOrderFieldError -from fastapi_toolsets.schemas import OffsetPagination +from fastapi_toolsets.schemas import OffsetPagination, PaginationType from .conftest import ( Role, RoleCreate, RoleCrud, + RoleCursorCrud, + RoleRead, User, UserCreate, UserCrud, @@ -1193,3 +1195,245 @@ class TestOrderParamsSchema: assert results[0].username == "alice" assert results[1].username == "charlie" + + +class TestOffsetParamsSchema: + """Tests for AsyncCrud.offset_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() + 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" + + 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) + 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) + sig = inspect.signature(dep) + le = next( + m.le + for m in sig.parameters["items_per_page"].default.metadata + if hasattr(m, "le") + ) + assert le == 50 + + def test_include_total_not_a_query_param(self): + """include_total is not exposed as a query parameter.""" + dep = RoleCrud.offset_params() + 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 + ) + 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 + ) + 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) + 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.""" + await RoleCrud.create(db_session, RoleCreate(name="admin")) + dep = RoleCrud.offset_params() + 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 + + +class TestCursorParamsSchema: + """Tests for AsyncCrud.cursor_params().""" + + def test_returns_cursor_and_items_per_page_params(self): + """Returned dependency has cursor and items_per_page params.""" + dep = RoleCursorCrud.cursor_params() + 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" + + 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) + 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) + sig = inspect.signature(dep) + le = next( + m.le + for m in sig.parameters["items_per_page"].default.metadata + if hasattr(m, "le") + ) + assert le == 75 + + def test_cursor_defaults_to_none(self): + """cursor defaults to None.""" + dep = RoleCursorCrud.cursor_params() + 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() + 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.""" + await RoleCrud.create(db_session, RoleCreate(name="admin")) + dep = RoleCursorCrud.cursor_params() + params = await dep(cursor=None, items_per_page=10) + result = await RoleCursorCrud.cursor_paginate( + db_session, **params, schema=RoleRead + ) + assert result.pagination.items_per_page == 10 + + +class TestPaginateParamsSchema: + """Tests for AsyncCrud.paginate_params().""" + + def test_returns_all_params(self): + """Returned dependency has pagination_type, page, cursor, items_per_page (no include_total).""" + dep = RoleCursorCrud.paginate_params() + 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() + assert getattr(dep, "__name__") == "RolePaginateParams" + + def test_default_pagination_type(self): + """default_pagination_type is reflected in pagination_type default.""" + from fastapi_toolsets.schemas import PaginationType + + dep = RoleCursorCrud.paginate_params( + default_pagination_type=PaginationType.CURSOR + ) + sig = inspect.signature(dep) + assert ( + sig.parameters["pagination_type"].default.default == PaginationType.CURSOR + ) + + def test_default_page_size(self): + """default_page_size is reflected in items_per_page default.""" + dep = RoleCursorCrud.paginate_params(default_page_size=15) + 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) + sig = inspect.signature(dep) + le = next( + m.le + for m in sig.parameters["items_per_page"].default.metadata + if hasattr(m, "le") + ) + assert le == 60 + + def test_include_total_not_a_query_param(self): + """include_total is not exposed as a query parameter.""" + dep = RoleCursorCrud.paginate_params() + 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)( + pagination_type=PaginationType.OFFSET, + page=1, + cursor=None, + items_per_page=10, + ) + result_false = await RoleCursorCrud.paginate_params(include_total=False)( + pagination_type=PaginationType.OFFSET, + page=1, + cursor=None, + items_per_page=10, + ) + assert result_true["include_total"] is True + assert result_false["include_total"] is False + + @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() + result = await dep( + pagination_type=PaginationType.OFFSET, + page=2, + cursor=None, + items_per_page=10, + ) + assert result == { + "pagination_type": PaginationType.OFFSET, + "page": 2, + "cursor": None, + "items_per_page": 10, + "include_total": True, + } + + @pytest.mark.anyio + async def test_integrates_with_paginate_offset(self, db_session: AsyncSession): + """paginate_params output unpacks into paginate() for offset strategy.""" + from fastapi_toolsets.schemas import OffsetPagination + + await RoleCrud.create(db_session, RoleCreate(name="admin")) + params = await RoleCursorCrud.paginate_params()( + pagination_type=PaginationType.OFFSET, + page=1, + cursor=None, + items_per_page=10, + ) + result = await RoleCursorCrud.paginate(db_session, **params, schema=RoleRead) + assert isinstance(result.pagination, OffsetPagination) + + @pytest.mark.anyio + async def test_integrates_with_paginate_cursor(self, db_session: AsyncSession): + """paginate_params output unpacks into paginate() for cursor strategy.""" + from fastapi_toolsets.schemas import CursorPagination + + await RoleCrud.create(db_session, RoleCreate(name="admin")) + params = await RoleCursorCrud.paginate_params()( + pagination_type=PaginationType.CURSOR, + page=1, + cursor=None, + items_per_page=10, + ) + result = await RoleCursorCrud.paginate(db_session, **params, schema=RoleRead) + assert isinstance(result.pagination, CursorPagination)