Offset-free Paging for SQLAlchemy

Warnings

• breaking Python versions earlier than 3.8 are no longer supported. Users on older Python versions must upgrade or use an older `sqlakeyset` release.
  Fix: Upgrade your Python environment to 3.8 or newer.
• gotcha Keysets must be unique per row. Always include primary key column(s) at the end of your `order_by` clause to ensure uniqueness and prevent skipped or repeated rows.
  Fix: Modify your `order_by` clause to include the primary key(s) as the final ordering criteria, e.g., `query.order_by(..., MyModel.id)`.
• gotcha Rows containing `NULL` values in keyset columns will be omitted from results, as SQL comparisons against `NULL` are always false. Ordering columns should be `NOT NULL`.
  Fix: Ensure ordering columns are defined as `NOT NULL` in your schema, or use `sqlalchemy.func.coalesce()` to provide a non-NULL default value for nullable ordering columns in your query.
• gotcha The built-in keyset serialization currently handles only basic data/column types (strings, ints, floats, datetimes, dates, booleans). More advanced custom types may require extending the serialization.
  Fix: For complex types, consult the documentation (when available) or implement custom serialization logic for your bookmarks.
• deprecated The `get_page` function is for legacy SQLAlchemy 1.3 style ORM queries. For SQLAlchemy 2.0 style queries using `select()`, `select_page` should be used.
  Fix: Migrate to SQLAlchemy 2.0 style queries and use `select_page(session, select(MyModel).order_by(...), ...)`.

Install

• pip install sqlakeyset Install stable version

Imports

• select_page

  from sqlakeyset import select_page

  Use for SQLAlchemy 2.0 style queries with a `Session` or `Connection` object.
• get_page

  from sqlakeyset import get_page

  Use for legacy SQLAlchemy 1.3 ORM queries (which omit the session/connection argument).
• select_page (asyncio)

  from sqlakeyset.asyncio import select_page

  Use for asynchronous SQLAlchemy queries with `AsyncSession`.

Quickstart

Demonstrates basic keyset pagination using `select_page` with a SQLAlchemy 2.0-style query, retrieving the first page and its bookmark for subsequent pages. It includes a simple model, in-memory SQLite setup, data insertion, and then the pagination logic.

from sqlalchemy import create_engine, select, String, Integer
from sqlalchemy.orm import declarative_base, Session, Mapped, mapped_column
from sqlakeyset import select_page

# 1. Define Base
Base = declarative_base()

# 2. Define a simple model
class Book(Base):
    __tablename__ = "books"
    id: Mapped[int] = mapped_column(Integer, primary_key=True)
    author: Mapped[str] = mapped_column(String(50))
    title: Mapped[str] = mapped_column(String(100))

    def __repr__(self):
        return f"Book(id={self.id}, author='{self.author}', title='{self.title}')"

# 3. Create an in-memory SQLite engine
engine = create_engine("sqlite:///:memory:")

# 4. Create tables
Base.metadata.create_all(engine)

# 5. Insert some sample data
with Session(engine) as session:
    session.add_all([
        Book(author="Stephen King", title="It"),
        Book(author="J.R.R. Tolkien", title="The Hobbit"),
        Book(author="Stephen King", title="The Stand"),
        Book(author="Frank Herbert", title="Dune"),
        Book(author="J.R.R. Tolkien", title="The Lord of the Rings"),
    ])
    session.commit()

# 6. Perform keyset pagination
with Session(engine) as session:
    # Build a query with ordering, including a unique key (id) at the end
    q = select(Book).order_by(Book.author, Book.title, Book.id)

    # Get the first page
    page1 = select_page(session, q, per_page=2)
    print(f"Page 1 items: {page1.items}")
    print(f"Page 1 has next: {page1.paging.has_next}")
    print(f"Page 1 next bookmark: {page1.paging.bookmark_next}")

    # Get the second page using the bookmark from the first page
    if page1.paging.has_next:
        page2 = select_page(session, q, per_page=2, bookmark=page1.paging.bookmark_next)
        print(f"Page 2 items: {page2.items}")
        print(f"Page 2 has next: {page2.paging.has_next}")
        print(f"Page 2 next bookmark: {page2.paging.bookmark_next}")