Internal Documentation

This page documents functions and classes that are used internally in sqlakeyset.

sqlakeyset.paging

Main paging interface and implementation.

sqlakeyset.paging.compare_tuples(lesser: Sequence, greater: Sequence, dialect: Dialect | None = None) → ColumnElement[bool][source]

Given two sequences of equal length (whose entries can be SQL clauses or simple values), create an SQL clause defining the lexicographic tuple comparison lesser < greater.

If dialect is provided and is an sqlalchemy SQL dialect supporting native tuple comparison, the SQL emitted is a native tuple comparison. Otherwise it is built manually using OR and AND.

sqlakeyset.paging.core_get_page(s: Session | Connection, selectable: Select, per_page: int, place: Tuple | None, backwards: bool) → Page[Row[_TP]][source]

Get a page from an SQLAlchemy Core selectable.

Parameters:

s – sqlalchemy.engine.Connection or sqlalchemy.orm.session.Session to use to execute the query.
selectable – The source selectable.
per_page – Number of rows per page.
place – Keyset representing the place after which to start the page.
backwards – If True, reverse pagination direction.

Returns:

Page

sqlakeyset.paging.core_page_from_rows(paging_select: _PagingSelect, rows: Sequence, keys: List[str], result_type, page_size: int, backwards: bool = False, current_place: Tuple | None = None) → Page[Row][source]: Turn a raw page of results for an SQLAlchemy Core query (as obtained by core_get_page()) into a Page for external consumers.

sqlakeyset.paging.orm_get_page(q: Query, per_page: int, place: Tuple | None, backwards: bool) → Page[source]

Get a page from an SQLAlchemy ORM query.

Parameters:

q – The Query to paginate.
per_page – Number of rows per page.
place – Keyset representing the place after which to start the page.
backwards – If True, reverse pagination direction.

Returns:

Page

sqlakeyset.paging.orm_page_from_rows(paging_query: _PagingQuery, rows: Sequence[Row], keys: List[str], result_type, page_size: int, backwards: bool = False, current_place: Tuple | None = None) → Page[source]: Turn a raw page of results for an ORM query (as obtained by orm_get_page()) into a results.Page for external consumers.

sqlakeyset.paging.where_condition_for_page(ordering_columns: List[OC], place: Tuple, dialect: Dialect) → ColumnElement[bool][source]

Construct the SQL condition required to restrict a query to the desired page.

Parameters:

ordering_columns (list(columns.OC)) – The query’s ordering columns
place (tuple) – The starting position for the page
dialect – The SQL dialect in use

Returns:

An SQLAlchemy expression suitable for use in .where() or .filter().

sqlakeyset.columns

Classes and supporting functions to manipulate ordering columns and extract keyset markers from query results.

class sqlakeyset.columns.AppendedColumn(oc, name=None)[source]

An ordering key that requires an additional column to be added to the original query.

extra_column: ColumnElement: An extra SQLAlchemy ORM entity that this ordering column needs to add to its query in order to retrieve its value at each row. If no extra data is required, the value of this property will be None.

get_from_row(row)[source]: Extract the value of this ordering column from a result row.

property ob_clause: The original ORDER BY (sub)clause underlying this column.

class sqlakeyset.columns.AttributeColumn(oc, index, attr)[source]

An ordering key that was included as a column attribute in the original query.

get_from_row(row)[source]: Extract the value of this ordering column from a result row.

class sqlakeyset.columns.DirectColumn(oc, index)[source]

An ordering key that was directly included as a column in the original query.

get_from_row(row)[source]: Extract the value of this ordering column from a result row.

class sqlakeyset.columns.MappedOrderColumn(oc: OC)[source]

An ordering column in the context of a particular query/select.

This wraps an OC with one extra piece of information: how to retrieve the value of the ordering key from a result row. For some queries, this requires adding extra entities to the query; in this case, extra_column will be set.

extra_column: ColumnElement | None: An extra SQLAlchemy ORM entity that this ordering column needs to add to its query in order to retrieve its value at each row. If no extra data is required, the value of this property will be None.

abstract get_from_row(internal_row)[source]: Extract the value of this ordering column from a result row.

property ob_clause: The original ORDER BY (sub)clause underlying this column.

property reversed: A MappedOrderColumn representing the same column in the reversed order.

class sqlakeyset.columns.OC(x)[source]

Wrapper class for ordering columns; i.e. instances of sqlalchemy.sql.expression.ColumnElement appearing in the ORDER BY clause of a query we are paging.

property comparable_value: The ordering column/SQL expression in a form that is suitable for incorporating in a ROW(...) > ROW(...) comparision; i.e. with ordering modifiers and labels removed.

property element: ColumnElement: The ordering column/SQL expression with ordering modifier removed.

property is_ascending: Returns True if this column is ascending, False if descending.

pair_for_comparison(value, dialect)[source]

Return a pair of SQL expressions representing comparable values for this ordering column and a specified value.

Parameters:

value – A value to compare this column against.
dialect – The sqlalchemy.engine.interfaces.Dialect in use.

Returns:

A pair (a, b) such that the comparison a < b is the condition for the value of this OC being past value in the paging order.

property reversed: An OC representing the same column ordering, but reversed.

sqlakeyset.columns.derive_order_key(ocol, desc, index)[source]

Attempt to derive the value of ocol from a query column.

Parameters:

ocol – The OC to look up.
desc – Either a column description as in sqlalchemy.orm.query.Query.column_descriptions, or a sqlalchemy.sql.expression.ColumnElement.

Returns:

Either a MappedOrderColumn or None.

sqlakeyset.columns.find_order_key(ocol: OC, column_descriptions) → MappedOrderColumn[source]

Return a MappedOrderColumn describing how to populate the ordering column ocol from a query returning columns described by column_descriptions.

Parameters:

ocol – The OC to look up.
column_descriptions – The list of columns from which to attempt to derive the value of ocol.

Returns:

A MappedOrderColumn wrapping ocol.

sqlakeyset.columns.parse_ob_clause(selectable) → List[OC][source]: Parse the ORDER BY clause of a selectable into a list of OC instances.

sqlakeyset.columns.strip_labels(el: ColumnElement) → ColumnElement[source]: Remove labels from a sqlalchemy.sql.expression.ColumnElement.