Skip to content

pagination

CLASS DESCRIPTION
PaginationData

Metadata describing the state of a paginated query.
PaginationResponse

Paginated response container including both pagination metadata and result
FUNCTION DESCRIPTION
get_paginated_response

Execute a paginated query and return a structured response.
get_pagination_data

Apply pagination to a SQLAlchemy statement and compute pagination metadata.

Classes

PaginationData pydantic-model

Bases: BaseModel

Metadata describing the state of a paginated query.

ATTRIBUTE DESCRIPTION
current_page

TYPE: int

page_size

TYPE: int

total_count

TYPE: int

Fields:

  • current_page (int)
  • page_size (int)
  • total_count (int)
Source code in fractal_server/app/routes/pagination.py 
47
48
49
50
51
52
53
54
55
56
57
58
59
class PaginationData(BaseModel):
    """
    Metadata describing the state of a paginated query.

    Attributes:
        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 pydantic-model

Bases: PaginationData, Generic[T]

Paginated response container including both pagination metadata and result items.

Note: We include arbitrary_types_allowed=True for cases like 

1
PaginationResponse[X]
where X is an ORM table class (e.g. PaginationResponse[JobV2]). Not doing so would make openapi generation fail.

ATTRIBUTE DESCRIPTION
current_page

TYPE: int

page_size

TYPE: int

total_count

TYPE: int

items

TYPE: list[T]

Config:

  • arbitrary_types_allowed: True

Fields:

  • current_page (int)
  • page_size (int)
  • total_count (int)
  • items (list[T])
Source code in fractal_server/app/routes/pagination.py 
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
class PaginationResponse(PaginationData, Generic[T]):
    """
    Paginated response container including both pagination metadata and result
    items.

    Note: We include `arbitrary_types_allowed=True` for cases like
    ```
    PaginationResponse[X]
    ```
    where `X` is an ORM table class (e.g. `PaginationResponse[JobV2]`).
    Not doing so would make openapi generation fail.

    Attributes:
        current_page:
        page_size:
        total_count:
        items:
    """

    model_config = ConfigDict(arbitrary_types_allowed=True)
    items: list[T]

Functions:

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: Select[T]

stm_count

TYPE: Select[int]

pagination

TYPE: PaginationRequest

db

TYPE: AsyncSession

Source code in fractal_server/app/routes/pagination.py 
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
async def get_paginated_response(
    *,
    stm: Select[T],
    stm_count: Select[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[Unpack[Ts]]

stm_count

TYPE: Select[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 
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
async def get_pagination_data(
    *,
    stm: Select[Unpack[Ts]],
    stm_count: Select[int],
    pagination: PaginationRequest,
    db: AsyncSession,
) -> tuple[Select[Unpack[Ts]], 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_one()

    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,
        ),
    )