pagination

PaginationData

Bases: BaseModel

Metadata describing the state of a paginated query.

PARAMETER	DESCRIPTION
current_page
page_size
total_count

Source code in fractal_server/app/routes/pagination.py

class PaginationData(BaseModel):
    """
    Metadata describing the state of a paginated query.

    Args:
        current_page:
        page_size:
        total_count:
    """

    current_page: int = Field(ge=1)
    page_size: int = Field(ge=0)
    total_count: int = Field(ge=0)

PaginationResponse

Bases: PaginationData, Generic[T]

Paginated response container including both pagination metadata and result items.

PARAMETER	DESCRIPTION
current_page
page_size
total_count
items

Source code in fractal_server/app/routes/pagination.py

class PaginationResponse(PaginationData, Generic[T]):
    """
    Paginated response container including both pagination metadata and result
    items.

    Args:
        current_page:
        page_size:
        total_count:
        items:

    """

    items: list[T]

get_paginated_response(*, stm, stm_count, pagination, db) async

Execute a paginated query and return a structured response.

This only applies to SelectOfScalar[T] statements, i.e. applies to select(X) but not to select(X, Y).

PARAMETER	DESCRIPTION
stm	TYPE: SelectOfScalar[T]
stm_count	TYPE: SelectOfScalar[int]
pagination	TYPE: PaginationRequest
db	TYPE: AsyncSession

Source code in fractal_server/app/routes/pagination.py

async def get_paginated_response(
    *,
    stm: SelectOfScalar[T],
    stm_count: SelectOfScalar[int],
    pagination: PaginationRequest,
    db: AsyncSession,
) -> PaginationResponse[T]:
    """
    Execute a paginated query and return a structured response.

    This only applies to `SelectOfScalar[T]` statements, i.e. applies to
    `select(X)` but not to `select(X, Y)`.

    Args:
        stm:
        stm_count:
        pagination:
        db:
    """
    stm, pagination_data = await get_pagination_data(
        stm=stm,
        stm_count=stm_count,
        pagination=pagination,
        db=db,
    )

    res = await db.execute(stm)
    records = res.scalars().all()

    return PaginationResponse[T](items=records, **pagination_data.model_dump())

get_pagination_data(*, stm, stm_count, pagination, db) async

Apply pagination to a SQLAlchemy statement and compute pagination metadata.

This function executes a separate count query to determine the total number of available items, then applies the appropriate OFFSET and LIMIT to the provided statement based on the requested pagination parameters.

PARAMETER	DESCRIPTION
stm	TYPE: Select[T] | SelectOfScalar[T]
stm_count	TYPE: SelectOfScalar[int]
pagination	TYPE: PaginationRequest
db	TYPE: AsyncSession

Returns: A tuple containing: - The modified SQLAlchemy statement with proper OFFSET and LIMIT. - A PaginationData instance with: * current_page: the requested page number; * page_size: the effective page size; * total_count: the total number of available items.

Source code in fractal_server/app/routes/pagination.py

async def get_pagination_data(
    *,
    stm: Select[T] | SelectOfScalar[T],
    stm_count: SelectOfScalar[int],
    pagination: PaginationRequest,
    db: AsyncSession,
) -> tuple[Select[T] | SelectOfScalar[T], PaginationData]:
    """
    Apply pagination to a SQLAlchemy statement and compute pagination metadata.

    This function executes a separate count query to determine the total number
    of available items, then applies the appropriate OFFSET and LIMIT to the
    provided statement based on the requested pagination parameters.

    Args:
        stm:
        stm_count:
        pagination:
        db:
    Returns:
        A tuple containing:
            - The modified SQLAlchemy statement with proper OFFSET and LIMIT.
            - A `PaginationData` instance with:
                * current_page: the requested page number;
                * page_size: the effective page size;
                * total_count: the total number of available items.
    """

    res_total_count = await db.execute(stm_count)
    total_count = res_total_count.scalar()

    if pagination.page_size is not None:
        page_size = pagination.page_size
        stm = stm.offset((pagination.page - 1) * page_size).limit(page_size)
    else:
        page_size = total_count

    return (
        stm,
        PaginationData(
            current_page=pagination.page,
            page_size=page_size,
            total_count=total_count,
        ),
    )