### Install litestar-pagination Source: https://github.com/zylvext/litestar-pagination/blob/main/docs/index.md Install the base package along with optional extras for SQLAlchemy or Advanced Alchemy support. ```bash pip install litestar-pagination pip install 'litestar-pagination[sqlalchemy]' pip install 'litestar-pagination[advanced-alchemy]' ``` -------------------------------- ### Execute sync and async pagination Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Usage examples for the paginate and apaginate helper functions. ```python page = paginate(session, statement, pagination) page = await apaginate(async_session, statement, pagination) ``` -------------------------------- ### SQLAlchemy Keyset Pagination Quickstart Source: https://github.com/zylvext/litestar-pagination/blob/main/docs/index.md Register CursorParams as a Litestar dependency and use apaginate to handle paginated database queries. ```python from litestar import Litestar, get from litestar.di import NamedDependency, Provide from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from litestar_pagination import CursorPage, CursorParams from litestar_pagination.ext.sqlalchemy import apaginate @get("/users") async def list_users( session: AsyncSession, pagination: NamedDependency[CursorParams], ) -> CursorPage[User]: statement = select(User).order_by(User.created_at, User.id) return await apaginate(session, statement, pagination) app = Litestar( route_handlers=[list_users], dependencies={"pagination": Provide(CursorParams, sync_to_thread=False)}, ) ``` -------------------------------- ### Initialize and verify the development environment Source: https://github.com/zylvext/litestar-pagination/blob/main/docs/contributing.md Run these commands to prepare the local environment and execute the test suite. ```bash just setup just check just test ``` -------------------------------- ### Import SQLAlchemy pagination utilities Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Import sync and async paginator classes and helper functions for SQLAlchemy. ```python from litestar_pagination.ext.sqlalchemy import ( SQLAlchemyAsyncCursorPaginator, SQLAlchemySyncCursorPaginator, apaginate, paginate, ) ``` -------------------------------- ### Run project verification and linting Source: https://github.com/zylvext/litestar-pagination/blob/main/AGENTS.md Executes linting, formatting, type checking, dependency analysis, and the test suite. ```bash uv run ruff check --fix . uv run ruff format . uv run ty check uv run deptry . just test ``` -------------------------------- ### Project Module Boundary Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md The required directory structure for the package, separating core cursor logic from storage-specific extensions. ```text litestar_pagination/ __init__.py cursor.py ext/ __init__.py sqlalchemy.py ``` -------------------------------- ### Implement framework-native cursor pagination Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Returns a CursorPagination response directly using the SQLAlchemyAsyncCursorPaginator for manual control over the pagination process. ```python from litestar.pagination import CursorPagination @get("/users") async def list_users( session: AsyncSession, pagination: NamedDependency[CursorParams], ) -> CursorPagination[str, User]: paginator = SQLAlchemyAsyncCursorPaginator( session, select(User).order_by(User.created_at, User.id), ) return await paginator(pagination.cursor, pagination.size) ``` -------------------------------- ### Import core pagination types Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Import the CursorPage and CursorParams dataclasses for use in your application. ```python from litestar_pagination import CursorPage, CursorParams ``` -------------------------------- ### Paginate with Advanced Alchemy Session Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Demonstrates using an injected AsyncSession and CursorParams dependency to paginate a SQLAlchemy select statement. ```python @get("/users") async def list_users( db_session: AsyncSession, pagination: NamedDependency[CursorParams], ) -> CursorPage[User]: return await apaginate( db_session, select(User).order_by(User.created_at, User.id), pagination, ) ``` -------------------------------- ### Integrate pagination with Litestar dependency injection Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Uses NamedDependency to inject CursorParams into a route handler and the apaginate helper to return a CursorPage. ```python from litestar import Litestar, get from litestar.di import NamedDependency, Provide from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from litestar_pagination import CursorPage, CursorParams from litestar_pagination.ext.sqlalchemy import apaginate @get("/users") async def list_users( session: AsyncSession, pagination: NamedDependency[CursorParams], ) -> CursorPage[User]: statement = select(User).order_by(User.created_at, User.id) return await apaginate(session, statement, pagination) app = Litestar( route_handlers=[list_users], dependencies={ "pagination": Provide(CursorParams, sync_to_thread=False), }, ) ``` -------------------------------- ### paginate(session, statement, pagination) Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Synchronous pagination function for SQLAlchemy sessions. ```APIDOC ## paginate(session, statement, pagination) ### Description Performs synchronous cursor-based pagination on a SQLAlchemy query. ### Parameters - **session** (SQLAlchemy Session) - Required - A compatible sync SQLAlchemy session. - **statement** (Select) - Required - A SQLAlchemy 2 Select statement selecting exactly one mapped ORM entity. - **pagination** (CursorParams) - Required - The pagination parameters. - **count_query** (Select) - Optional - An optional caller-supplied count query. - **unique** (bool) - Optional - A unique option defaulting to True. ### Returns - **CursorPage[T]** - Returns a rich bidirectional CursorPage object. ``` -------------------------------- ### Define CursorPage response model Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Conceptual generic response model for paginated results. ```python T = TypeVar("T") @dataclass class CursorPage(Generic[T]): items: Sequence[T] total: int | None current_page: str | None current_page_backwards: str | None previous_page: str | None next_page: str | None ``` -------------------------------- ### SQLAlchemy Native Forward-Only Pagination Source: https://github.com/zylvext/litestar-pagination/blob/main/docs/index.md Implementation of forward-only pagination using SQLAlchemyAsyncCursorPaginator, which avoids executing a total count query. ```python from litestar import get from litestar.di import NamedDependency from litestar.pagination import CursorPagination from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from litestar_pagination import CursorParams from litestar_pagination.ext.sqlalchemy import SQLAlchemyAsyncCursorPaginator @get("/users") async def list_users( session: AsyncSession, pagination: NamedDependency[CursorParams], ) -> CursorPagination[str, User]: paginator = SQLAlchemyAsyncCursorPaginator( session, select(User).order_by(User.created_at, User.id), ) return await paginator(pagination.cursor, pagination.size) ``` -------------------------------- ### apaginate(async_session, statement, pagination) Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Asynchronous pagination function for SQLAlchemy sessions. ```APIDOC ## apaginate(async_session, statement, pagination) ### Description Performs asynchronous cursor-based pagination on a SQLAlchemy query. ### Parameters - **async_session** (AsyncSession) - Required - A compatible async SQLAlchemy session. - **statement** (Select) - Required - A SQLAlchemy 2 Select statement selecting exactly one mapped ORM entity. - **pagination** (CursorParams) - Required - The pagination parameters. - **count_query** (Select) - Optional - An optional caller-supplied count query. - **unique** (bool) - Optional - A unique option defaulting to True. ### Returns - **CursorPage[T]** - Returns a rich bidirectional CursorPage object. ``` -------------------------------- ### Define CursorParams model Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md Conceptual model for cursor parameters using Litestar query parameter annotations. ```python @dataclass class CursorParams: cursor: Annotated[str | None, QueryParameter(name="cursor", required=False)] = None size: Annotated[int, QueryParameter(name="size", ge=1, le=100, required=False)] = 50 include_total: ClassVar[bool] = True ``` -------------------------------- ### Refresh GitHub Actions pins Source: https://github.com/zylvext/litestar-pagination/blob/main/AGENTS.md Lists pinned GitHub Actions and checks for updates or discrepancies. ```bash python3 .agents/skills/github-actions-pin-refresh/scripts/list_pinned_actions.py --check rg "uses:" .github -n git diff -- .github ``` -------------------------------- ### Define Deterministic Ordering Source: https://github.com/zylvext/litestar-pagination/blob/main/docs/index.md Ensure queries include a unique tie-breaker in the ORDER BY clause to satisfy pagination requirements. ```python select(User).order_by(User.created_at, User.id) ``` -------------------------------- ### Integrate pagination with Advanced Alchemy DTOs Source: https://github.com/zylvext/litestar-pagination/blob/main/docs/index.md Use SQLAlchemyDTO to transform entities within a paginated response by injecting the session directly into the handler. ```python from advanced_alchemy.extensions.litestar.dto import SQLAlchemyDTO from litestar import get from litestar.di import NamedDependency from sqlalchemy import select from sqlalchemy.ext.asyncio import AsyncSession from litestar_pagination import CursorPage, CursorParams from litestar_pagination.ext.sqlalchemy import apaginate @get("/users", return_dto=SQLAlchemyDTO[User]) async def list_users( db_session: AsyncSession, pagination: NamedDependency[CursorParams], ) -> CursorPage[User]: return await apaginate( db_session, select(User).order_by(User.created_at, User.id), pagination, ) ``` -------------------------------- ### CursorPage JSON Response Structure Source: https://github.com/zylvext/litestar-pagination/blob/main/docs/index.md The structure returned by paginate() and apaginate() methods, providing bidirectional navigation and total counts. ```json { "items": [{"id": 42, "name": "Ada"}], "total": 100, "current_page": "...", "current_page_backwards": "...", "previous_page": null, "next_page": "..." } ``` -------------------------------- ### Define a custom count query for pagination Source: https://github.com/zylvext/litestar-pagination/blob/main/docs/index.md Provide an explicit count query when the default subquery approach is insufficient for complex joins or filtered results. ```python from sqlalchemy import func, select count_query = select(func.count()).select_from(User).where(User.is_active) page = await apaginate(session, statement, pagination, count_query=count_query) ``` -------------------------------- ### Disable total count in pagination response Source: https://github.com/zylvext/litestar-pagination/blob/main/docs/index.md Subclass the dependency contract and set include_total to False to skip the extra count query, resulting in a null total field. ```python from typing import ClassVar class NoTotalCursorParams(CursorParams): include_total: ClassVar[bool] = False app = Litestar( route_handlers=[list_users], dependencies={"pagination": Provide(NoTotalCursorParams, sync_to_thread=False)}, ) ``` -------------------------------- ### Pagination Response JSON Structure Source: https://github.com/zylvext/litestar-pagination/blob/main/PRD.md The standard response format for paginated results. Cursor values are opaque and should not be parsed by the client. ```json { "items": [ {"id": 42, "name": "Ada"} ], "total": 100, "current_page": "...", "current_page_backwards": "...", "previous_page": null, "next_page": "..." } ``` === COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.