Sling Academy
Home/FastAPI/FastAPI + SQLAlchemy: Using cursor-based pagination

FastAPI + SQLAlchemy: Using cursor-based pagination

Last updated: February 22, 2024

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.

Previous Article: How to Implement Redirects in FastAPI (2 Ways)

Series: FastAPI Request & Response Tutorials

FastAPI

You May Also Like

  • Popular useful built-in Jinja filters you should know
  • How to remove consecutive whitespace in rendered Jinja pages
  • How to format large numbers with thousand separators in Jinja template
  • How to format date time in Jinja templates
  • FastAPI + Jinja: How to create custom filters
  • How to pass variables from Python (FastAPI) to Jinja
  • How to decode Jinja response to string
  • How to create and use macros in Jinja
  • How to use namespace in Jinja
  • How to use if/ else in Jinja
  • How to use loops in Jinja
  • FastAPI: How to use macros in Jinja templates
  • Fixing Common Swagger UI Errors in FastAPI
  • FastAPI Error: 307 Temporary Redirect – Causes and Solutions
  • FastAPI Error: Expected UploadFile, received ‘str’
  • Resolving FastAPI ImportError: No Known Parent Package
  • FastAPI Error: No module named ‘pydantic_core._pydantic_core’
  • Resolving FastAPI 422 Error: Value is not a valid dict
  • Resolving the FastAPI Circular References Error