Pagination

Introduction

When dealing with large datasets, it's often impractical to load all records at once. Pagination allows you to fetch a subset of records, making data retrieval more efficient and user-friendly. Expanse provides built-in support for pagination of SQLAlchemy queries in an intuitive way which does not require any configuration by default.

Expanse provides two types of pagination out of the box: offset-based and cursor-based pagination.

Offset-based pagination

Offset-based pagination is the most common type of pagination. It uses an offset and a limit to fetch a subset of records.

To start using offset-based pagination, you can use the paginate method on a query object. It will automatically configure the query's offset and limit based on the request's query parameters. By default, it uses the page query parameter to detect the current page.

from typing import Annotated
from expanse.database.session import Session
from expanse.pagination.paginator import Paginator
from pydantic import BaseModel
from sqlalchemy import select

from app.models.user import User


class UserSchema(BaseModel):
    id: int
    name: str
    email: str


class UserController:
    def index(
        self, session: Session
    ) -> Paginator[Annotated[User, UserSchema]]:
        paginator = session.paginate(
            select(User).order_by(User.id.desc()),
            per_page=20
        )

        return paginator

from typing import Annotated
from expanse.database.session import AsyncSession
from expanse.pagination.paginator import Paginator
from pydantic import BaseModel
from sqlalchemy import select

from app.models.user import User


class UserSchema(BaseModel):
    id: int
    name: str
    email: str


class UserController:
    async def index(
        self, session: AsyncSession
    ) -> Paginator[Annotated[User, UserSchema]]:
        paginator = await session.paginate(
            select(User).order_by(User.id.desc()),
            per_page=20
        )

        return paginator

In this example, we only need to specify the per_page parameter to define how many records we want to fetch per page, Expanse will take care of the rest, which includes the proper serialization of the paginated response.

{
  "data": [
    {
      "id": 100,
      "name": "User 100",
      "email": "[email protected]"
    }
    // ...
  ],
  "total": 250,
  "per_page": 20,
  "current_page": 1,
  "last_page": 13,
  "links": {
    "first": "http://example.com/users?page=1",
    "last": "http://example.com/users?page=13",
    "prev": null,
    "next": "http://example.com/users?page=2",
    "self": "http://example.com/users?page=1"
  }
}

This is the default, envelope-style response for paginated data. An alternative, header-based response is also supported.

The page the paginator needs to fetch is determined by the page query parameter in the request URL. For example, to fetch the second page of results, you would make a request to:

http://example.com/users?page=2

Controlling the paginated response

By default, paginators are serialized using a thorough, envelope-style response. However, you can customize the response format to make it leaner or by using an alternative, header-based response.

Envelope-based response

The default envelope response includes a complete set of pagination metadata, such as total records, current page, last page, etc. However, you might not need all this information in every response. You can customize the envelope response by annotating the Paginator with the Envelope variant.

from expanse.pagination.offset.adapters.envelope import Envelope


def index(
    self, session: Session
) -> Annotated[
    Paginator[Annotated[User, UserSchema]],
    Envelope(with_links=False)
]:
    ...

Here are the available options for the Envelope adapter:

with_links: Whether to include pagination links in the response. Default is True.

Header-based response

In some cases, you might prefer to include pagination metadata in HTTP headers instead of the response body. You can achieve this by using the Header adapter.

from expanse.pagination.offset.adapters.header import Header

def index(
    self, session: Session
) -> Annotated[
    Paginator[Annotated[User, UserSchema]],
    Header()
]:
    ...

When using the Header adapter, the response body will contain only the paginated data as a list of items, while pagination metadata will be included in the HTTP headers. Here are the headers that will be included in the response:

HTTP/1.1 200 OK
Content-Type: application/json
X-Total-Count: 250
Link: <http://example.com/users?page=1>; rel="first",
      <http://example.com/users?page=13>; rel="last",
      <http://example.com/users?page=2>; rel="next",
      <http://example.com/users?page=1>; rel="self"

[
  {
    "id": 100,
    "name": "User 100",
    "email": "[email protected]"
  }
  // ...
]

X-Total-Count: The otal number of records.
Link: Pagination links for the first (first), the last (last), the previous (prev), the next (next), and the current (self) pages.

Cursor-based pagination

Cursor-based pagination is an alternative to offset-based pagination that uses a cursor to keep track of the current position in the dataset. A cursor is converted to a WHERE clause to fetch the next set of records instead of an OFFSET clause to skip records. This is the most efficient way to paginate large datasets, and is particularly suited for infinite scrolling interfaces.

To start using cursor-based pagination, you can use the cursor_paginate method on a query object.

from typing import Annotated
from expanse.database.session import Session
from expanse.pagination.cursor.paginator import CursorPaginator
from pydantic import BaseModel
from sqlalchemy import select

from app.models.user import User

class UserSchema(BaseModel):
    id: int
    name: str
    email: str

class UserController:
    def index(
        self, session: Session
    ) -> CursorPaginator[Annotated[User, UserSchema]]:
        paginator = session.cursor_paginate(
            select(User).order_by(User.id.desc()),
            per_page=20,
        )

        return paginator

from typing import Annotated
from expanse.database.session import AsyncSession
from expanse.pagination.cursor.paginator import CursorPaginator
from pydantic import BaseModel
from sqlalchemy import select

from app.models.user import User

class UserSchema(BaseModel):
    id: int
    name: str
    email: str

class UserController:
    async def index(
        self, session: AsyncSession
    ) -> CursorPaginator[Annotated[User, UserSchema]]:
        paginator = await session.cursor_paginate(
            select(User).order_by(User.id.desc()),
            per_page=20,
        )

        return paginator

{
  "data": [
    {
      "id": 100,
      "name": "User 100",
      "email": "[email protected]"
    }
    // ...
  ],
  "next_cursor": "eyJwYXJhbWV0ZXJzIjogeyJpZCI6IDJ9LCAicmV2ZXJzZWQiOiBmYWxzZX0=",
  "previous_cursor": null,
  "links": {
    "next": "http://example.com/users?cursor=eyJwYXJhbWV0ZXJzIjogeyJpZCI6IDJ9LCAicmV2ZXJzZWQiOiBmYWxzZX0=",
    "prev": null,
    "self": "http://example.com/users"
  }
}

This is the default, envelope-style response for cursor-paginated data. An alternative, header-based response is also supported.

The cursor the paginator needs to fetch is determined by the cursor query parameter in the request URL. For example, to fetch the next page of results, you would make a request to:

http://example.com/users?cursor=eyJwYXJhbWV0ZXJzIjogeyJpZCI6IDJ9LCAicmV2ZXJzZWQiOiBmYWxzZX0=

Cursor-based pagination requires that the query has a deterministic ordering. Make sure to include an order_by clause in your query and that it uniquely identifies each record.

Also, the ordering columns must belong to the table being paginated.

Controlling the paginated response

By default, paginators are serialized using a thorough, envelope-style response. However, you can customize the response format to use an alternative, header-based response.

Header-based response

In some cases, you might prefer to include pagination metadata in HTTP headers instead of the response body. You can achieve this by using the Header adapter.

from expanse.pagination.cursor.adapters.header import Header

def index(
    self, session: Session
) -> Annotated[
    CursorPaginator[Annotated[User, UserSchema]],
    Header()
]:
    ...

HTTP/1.1 200 OK
Content-Type: application/json
X-Total-Count: 250
Link: <http://example.com/users?cursor=eyJwYXJhbWV0ZXJzIjogeyJpZCI6IDJ9LCAicmV2ZXJzZWQiOiBmYWxzZX0%30>; rel="next",
      <http://example.com/users>; rel="self"

[
  {
    "id": 100,
    "name": "User 100",
    "email": "[email protected]"
  }
  // ...
]

Link: Pagination links for the previous (prev), the next (next), and the current (self) pages.