sqlalchemy-relay-pagination

Python package that make setting up relay-style pagination with SQLAlchemy easy.

Quickstart

Installation

uv add sqlalchemy-relay-pagination

Schema

Define your connection types in SDL:

type Query {
  users(first: Int, last: Int, after: String, before: String): UserConnection!
}

type UserConnection {
  edges: [UserEdge!]!
  pageInfo: PageInfo!
}

type UserEdge {
  node: User!
  cursor: String!
}

type PageInfo {
  hasNextPage: Boolean!
  hasPreviousPage: Boolean!
  startCursor: String
  endCursor: String
  totalCount: Int
}

type User {
  id: Int!
  name: String!
  posts(first: Int, last: Int, after: String, before: String): PostConnection!
}

type PostConnection {
  edges: [PostEdge!]!
  pageInfo: PageInfo!
}

type PostEdge {
  node: Post!
  cursor: String!
}

type Post {
  id: Int!
  title: String!
}

Resolver

from ariadne import QueryType
from sqlalchemy import select
from sqlalchemy_relay_pagination import paginate
from sqlalchemy_relay_pagination.graphql import extract_requested_fields

query_type = QueryType()

@query_type.field("users")
async def resolve_users(_, info, first=None, last=None, after=None, before=None):
    session = info.context["session"]
    return await paginate(
        session,
        select(User),
        order_by=User.id,
        first=first,
        last=last,
        after=after,
        before=before,
        requested_fields=extract_requested_fields(info),
    )

extract_requested_fields inspects the GraphQL selection set and tells paginate which fields the client actually requested — skipping the COUNT(*) query when totalCount is absent, and skipping the extra-row fetch when neither hasNextPage nor hasPreviousPage is selected.

Batch-loading nested connections (avoiding N+1)

For nested paginated fields (e.g. User.posts), use paginate_many inside a dataloader to fetch all parents' children in a single query:

from aiodataloader import DataLoader
from ariadne import ObjectType
from sqlalchemy_relay_pagination import paginate_many

user_type = ObjectType("User")

class UserPostsLoader(DataLoader):
    def __init__(self, session, first, last, fields):
        super().__init__()
        self._session = session
        self._first = first
        self._last = last
        self._fields = fields

    async def batch_load_fn(self, user_ids):
        results = await paginate_many(
            self._session,
            select(Post),
            model=Post,
            partition_by=Post.user_id,
            order_by=Post.id,
            keys=list(user_ids),
            first=self._first,
            last=self._last,
            requested_fields=self._fields,
        )
        return [results[uid] for uid in user_ids]

@user_type.field("posts")
async def resolve_user_posts(user, info, first=None, last=None):
    session = info.context["session"]
    loaders = info.context["loaders"]

    loader_key = (first, last)
    if loader_key not in loaders:
        loaders[loader_key] = UserPostsLoader(
            session, first, last, extract_requested_fields(info)
        )

    return await loaders[loader_key].load(user.id)

Note: when after or before cursors are provided, paginate_many falls back to one query per parent key instead of a single batched query, because each parent may be at a different cursor position. Results are still correct; only the single-query optimisation is lost.

Global configuration

Call configure() once at application startup:

from sqlalchemy_relay_pagination import configure, PaginationConfig

configure(PaginationConfig(default_page_size=20, max_page_size=100))

Settings can also be provided via environment variables:

SQLALCHEMY_RELAY_PAGINATION_DEFAULT_PAGE_SIZE=20
SQLALCHEMY_RELAY_PAGINATION_MAX_PAGE_SIZE=100

Resources

• GraphQL pagination documentation page
• Apollo's Relay-style pagination overview
• GraphQL Cursor Connections Specification
• GraphQL Cursor Connections: Filtering and Ordering