Litestar Pagination

https://github.com/zylvext/litestar-pagination
Admin
Litestar Pagination is a library providing cursor/keyset pagination for Litestar, with optional...
Tokens:5,412
Snippets:78
Trust Score:4.1
License:MIT
Update:3 months ago
Show doc for...
Tokens:
Raw
### 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": "..."
}
```
Tokens:
Raw