DB Pagination

The Pagination class is a tool for managing paginated rows from the database.

It requires the db extra to be installed:

pip install aio-fluid[db]

It can be imported from fluid.db:

from fluid.db import Pagination, Search

fluid.db.Pagination

Bases: NamedTuple

order_by_fields instance-attribute

order_by_fields

Fields to order results by

limit instance-attribute

limit

Maximum number of results per page

filters instance-attribute

filters

Filters applied to the query

search instance-attribute

search

Full-text search configuration

cursor instance-attribute

cursor

Decoded pagination cursor

desc class-attribute instance-attribute

desc = False

Order results in descending order

order_by_fields_sign property

order_by_fields_sign

create classmethod

create(
    *order_by_fields,
    cursor="",
    limit=None,
    filters=None,
    search=None,
    desc=False
)

Factory method to create a Pagination instance, decoding the cursor if provided.

If the cursor is provided, filters, limit and search are extracted from it, and the provided values for these parameters are ignored.

PARAMETER	DESCRIPTION
*order_by_fields	Fields to order results by TYPE: str DEFAULT: ()
cursor	Encoded pagination cursor from a previous response TYPE: str DEFAULT: ''
limit	Maximum number of results per page; defaults to settings.DEFAULT_PAGINATION_LIMIT TYPE: int | None DEFAULT: None
filters	Filters to apply to the query TYPE: dict[str, Any] | None DEFAULT: None
search	Full-text search configuration TYPE: Search | None DEFAULT: None
desc	Order results in descending order TYPE: bool DEFAULT: False

Source code in fluid/db/pagination.py

@classmethod
def create(
    cls,
    *order_by_fields: Annotated[str, Doc("Fields to order results by")],
    cursor: Annotated[
        str,
        Doc("Encoded pagination cursor from a previous response"),
    ] = "",
    limit: Annotated[
        int | None,
        Doc(
            "Maximum number of results per page; "
            "defaults to settings.DEFAULT_PAGINATION_LIMIT"
        ),
    ] = None,
    filters: Annotated[
        dict[str, Any] | None,
        Doc("Filters to apply to the query"),
    ] = None,
    search: Annotated[
        Search | None,
        Doc("Full-text search configuration"),
    ] = None,
    desc: Annotated[
        bool,
        Doc("Order results in descending order"),
    ] = False,
) -> Self:
    """Factory method to create a Pagination instance,
    decoding the cursor if provided.

    If the cursor is provided, filters, limit and search are extracted from it,
    and the provided values for these parameters are ignored.
    """
    if cursor:
        decoded_cursor = Cursor.decode(cursor, order_by_fields)
        limit = decoded_cursor.limit
        filters = decoded_cursor.filters
        if search:
            search = search._replace(search_text=decoded_cursor.search_text)
    else:
        decoded_cursor = None
        limit = limit or settings.DEFAULT_PAGINATION_LIMIT
    return cls(
        order_by_fields=order_by_fields,
        cursor=decoded_cursor,
        limit=limit,
        filters=filters or {},
        search=search,
        desc=desc,
    )

execute async

execute(db, table, *, conn=None)

Execute the paginated query and return the results along with the next cursor.

PARAMETER	DESCRIPTION
db	Database instance to execute the query on TYPE: CrudDB
table	SQLAlchemy table to query TYPE: FromClause
conn	Optional existing connection to reuse TYPE: AsyncConnection | None DEFAULT: None

Source code in fluid/db/pagination.py

async def execute(
    self,
    db: Annotated[CrudDB, Doc("Database instance to execute the query on")],
    table: Annotated[FromClause, Doc("SQLAlchemy table to query")],
    *,
    conn: Annotated[
        AsyncConnection | None,
        Doc("Optional existing connection to reuse"),
    ] = None,
) -> tuple[Sequence[Row], str]:
    """Execute the paginated query and return the results
    along with the next cursor.
    """
    sql_query = self.query(db, table)
    async with db.ensure_connection(conn) as conn:
        result = await conn.execute(sql_query)
    data = result.all()
    cursor = ""
    if self.limit > 0 and len(data) > self.limit:
        cursor = self._encode_cursor(data[-1])
        data = data[:-1]
    return data, cursor

query

query(db, table)

Source code in fluid/db/pagination.py

def query(self, db: CrudDB, table: FromClause) -> Select:
    sql_query = cast(
        Select,
        db.get_query(table, table.select(), params=self.filters),
    )
    if self.search:
        sql_query = db.search_query(
            table,
            sql_query,
            self.search.search_fields,
            self.search.search_text,
        )
    start_clause = self._start_clause(table)
    if start_clause is not None:
        sql_query = sql_query.where(start_clause)
    columns = db.order_by_columns(table, self.order_by_fields_sign)
    ordered = sql_query.order_by(*columns)
    return ordered.limit(self.limit + 1) if self.limit > 0 else ordered

fluid.db.Search

Bases: NamedTuple

search_fields instance-attribute

search_fields

Fields to search in

search_text instance-attribute

search_text

Text to search for