FastAPI + SQLAlchemy: Using cursor-based pagination

Introduction

This detailed tutorial introduces cursor-based pagination in applications using FastAPI with SQLAlchemy. We will journey from understanding the importance of pagination to implementing a cursor-based approach. This guide assumes you have basic knowledge of Python, FastAPI, and SQLAlchemy.

Understanding Pagination and Cursor-Based Pagination

Pagination is a technique used to divide large amounts of data into manageable chunks or pages. It enhances application performance by limiting the amount of data transferred with each request and improving user experience by providing structured data presentation. Cursor-based pagination, specifically, uses a pointer (cursor) to navigate through records, offering more consistent performance with large data sets and frequent updates.

Setting Up Your Environment

First, ensure you have FastAPI, SQLAlchemy, and a database driver (e.g., asyncpg for PostgreSQL) installed:

pip install fastapi sqlalchemy asyncpg uvicorn

For demonstration purposes, we’ll set up a simple FastAPI application with SQLAlchemy ORM:

from fastapi import FastAPI
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

app = FastAPI()

DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"

engine = create_async_engine(DATABASE_URL, echo=True)
AsyncSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine, class_=AsyncSession)

Base = declarative_base()

Next, we’ll create a very simple model and a route to expose it:

from sqlalchemy import Column, Integer, String

class Article(Base):
    __tablename__ = 'articles'
    id = Column(Integer, primary_key=True)
    title = Column(String)
    content = Column(String)

@app.get("/articles")
async def read_articles():
    # Imaginary function to fetch articles
    return []

Implementing Cursor-based Pagination

The core of cursor-based pagination lies in selecting a suitable column as a cursor, typically a unique, sequentially incrementing value such as an ID or a timestamp. For this example, we’ll use the ‘id’ column of our Article model.

We need to adjust our read_articles route to implement pagination:

@app.get("/articles")
async def read_articles(cursor: int = None):
    async with AsyncSessionLocal() as session:
        query = select(Article).order_by(Article.id)
        if cursor:
            query = query.filter(Article.id > cursor)
        results = await session.execute(query)
        articles = results.scalars().all()
        return {"data": articles, "next_cursor": articles[-1].id if articles else None}

In the above code, ‘cursor’ is a query parameter. If provided, it filters the dataset to return only records with an ID greater than the cursor. This effectively moves the ‘pagination window’ forward. To fetch the next ‘page’, clients use the ‘next_cursor’ returned from the last request.

Improving the Pagination Logic

While the basic implementation achieves pagination, you may wish to enhance it by:

• Limiting the number of results returned in each request.
• Adding previous page functionality using a ‘prev_cursor’.
• Implementing direction-aware pagination to move forwards or backwards.

These modifications would make the pagination feature more flexible and user-friendly.

Advanced Usage

For a more sophisticated application, you can combine cursor-based pagination with other features of SQLAlchemy and FastAPI such as:

• Dynamic filtering and sorting based on query parameters.
• Integration with FastAPI’s dependency injection system to streamline database session management.
• User authentication and authorization to control access to paginated resources.

Here’s a simplified example integrating dynamic sorting:

@app.get("/articles")
async def read_articles(cursor: int = None, order: str = 'asc'):
    async with AsyncSessionLocal() as session:
        query = select(Article)
        if order == 'desc':
            query = query.order_by(Article.id.desc())
        else:
            query = query.order_by(Article.id.asc())

        if cursor:
            filter_method = Article.id < cursor if order == 'desc' else Article.id > cursor
            query = query.filter(filter_method)

        results = await session.execute(query)
        articles = results.scalars().all()
        return {"data": articles, "next_cursor": articles[0].id if order == 'desc' and articles else articles[-1].id if articles else None}

Conclusion

Cursor-based pagination in FastAPI and SQLAlchemy setups is a potent method to handle large datasets with efficiency and user-friendly navigation. The basics provided in this tutorial equip you to implement and modify pagination features to suit diverse application needs.