How to implement pagination in FastAPI feat. SQLAlchemy

In this blog, I will explain how we can implement custom pagination in FastAPI without any external packages.

from typing import Generic, List, TypeVar
from pydantic import BaseModel, conint
from pydantic.generics import GenericModel

class PageParams(BaseModel):
    """ Request query params for paginated API. """
    page: conint(ge=1) = 1
    size: conint(ge=1, le=100) = 10


T = TypeVar("T")

class PagedResponseSchema(GenericModel, Generic[T]):
    """Response schema for any paged API."""

    total: int
    page: int
    size: int
    results: List[T]

First, we declare the pydantic schema for request and response. Here conint (constrained int) is used which does validation checks.

PagedResponseSchema is a generic pydantic model which takes any other pydantic model and puts it in results values as a list.

from fastapi import FastAPI, Depends, Request, Query
from pagination import PagedResponseSchema, PageParams, paginate
from pydantic import BaseModel
from sqlalchemy.orm import Session
from db import get_db
from user import User

app = FastAPI()

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

    class Config:
        orm_mode = True

@app.get("/users", response_model=PagedResponseSchema[UserSchema])
def get_users(request: Request, page_params: PageParams = Depends(), db: Session = Depends(get_db)):

    # # Add 1000 users to the database

    # for i in range(1000):
    #     db.add(User(id=i, name=f"User {i}"))

    # db.commit()

    query = db.query(User)

    return paginate(page_params, query, UserSchema)

Next we have our main.py file where our fastAPI server will be running.

In UserSchema we have added orm_mode: True which will take sql ORM model and converts it into pydantic schema. It also does many fancy things you can read in docs

After that one key point to note is how the page_params is declared. This way it is interpreted as query params and not as request body.

Now time for our special sause paginate function.

def paginate(page_params: PageParams, query, ResponseSchema: BaseModel) -> PagedResponseSchema[T]:
    """Paginate the query."""

    paginated_query = query.offset((page_params.page - 1) * page_params.size).limit(page_params.size).all()

    return PagedResponseSchema(
        total=query.count(),
        page=page_params.page,
        size=page_params.size,
        results=[ResponseSchema.from_orm(item) for item in paginated_query],
    )

This function is not that trivial, it just takes params and query and then adds offset and limit to that query. Now since ORM does lazy loading the call to the database has not happened yet that's why we can add limit and offset values.

And voila! Your performant pagination API is ready.

...

You can find the sample code on GitHub github.com/jayhawk24/pagination-fastapi . Feel free to follow me on GitHub

How to implement pagination in FastAPI feat. SQLAlchemy

Table of contents

No headings in the article.