Paginating data in Fast API

FastAPI is an amazing Python micro-framework. It can be used for creating serverless APIs, REST APIs, GraphQL APIs, and complete web applications.

Generally, if you are using an SQL database in your FastAPI app/service, you might end up using SQLAlchemy. You might need to send a good chunk of data and thus for better performance, you might need to cache your data. Now the problem is SQLAlchemy does not have out-of-the-box support for pagination. So how can we achieve pagination with Fast API and SQLAlchemy? Well, let's get into it.

PS: This post assumes that you have a basic understanding of FastAPI, Pydantic models, SQLAlchemy and how it works

First, let's create a simple Fast API app. Now I will be creating this as a REST API service. But before we dive into details, we need to make sure that the latest version of Python is installed correctly. We can check this using the following command in the terminal

> python --version

You should see the Python version as a result of the above command. Now if everything is good let's create our project.

In your terminal type this command in the project path where you want to create this project. For instance, if you are on Windows and want to create the project in the L drive pythonScripts folder then your path should be

L:\pythonScripts>

No, follow the steps mentioned here to create a new FastAPI project.

Once you create your main.py, create files as per the directory structure below. (ignore the __pycache__ folder)

Let's add some code in the database.py file

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

from sqlalchemy.ext.declarative import declarative_base

databasename="your database name"
username="your database username"
password="your database password"
host="your database host address"

postgresDBUrl=f"postgresql://{username}:{password}@{host}/{databasename}"

engine = create_engine(postgresDBUrl)

SessionLocal = sessionmaker(bind=engine, autoflush=False, autocommit=False)

Base = declarative_base()

First, we begin by importing create_engine and sesssionmaker from sqlalchemy and sqlalchemy.orm respectively. create_engine will create the database engine instance which will help SQLAlchemy ORM models and our database to talk with each other. Once the engine is created, we could use it to connect to the database and execute SQL queries.

We create the PostgreSQL connection URL using the database name, username, password, and host. Then we need to pass this URL to the create_engine to get the engine instance.

Now we will bind this engine to the database session using the sessionmaker.

Once done, we will need the Base class instance which we will get from declarative_base(). Our database models will be inheriting this Base class.

Now next thing we need to do is create our database models which are nothing but SQLAlchemy models (Python classes) which will map to a table in our database.

Inside models.py add the below code 

from tkinter.tix import INTEGER
from tokenize import String
import database as db
from sqlalchemy import Column, Integer, String

class User(db.Base):
    __tablename__="users"
    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50))
    email = Column(String(255))

Our User model will map to the users table in our database. For simplicity let's just have 3 columns in this table. This User model inherits the Base class that we declared in the database.py file.

It's better to keep our business/application logic in helper or service class/files. Let's create a user_crud.py helper file. Once created add the below code to it

import models

def getAllUsers(db):
    return db.query(models.User).all()

def createUser(user, db):
    new_user = models.User(name=user.name, email=user.email)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user

For simplicity user_crud.py file has two functions getAllUsers to retrieve all users and createUser to create a new user. The db instance passed to each of these functions will be our database session instance that we will pass from our main.py file. 

To see some random data in our table let's use Faker package. To install the package type the following command in your terminal

> pip install Faker

Once done, let's add a seeder function in our user_crud.py file. So our code should look as below

import models
from faker import Faker

def getAllUsers(db):
    return db.query(models.User).all()

def createUser(user, db):
    new_user = models.User(name=user.name, email=user.email)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


def seedUsers(db):
    fake = Faker()
    for i in range(40):
        user = models.User(name=fake.name(), email=fake.email())
        db.add(user)
        db.commit()

Now let's go to our main.py file and define our routes and required functions.

Here is the main.py file code

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from typing import List
import models
import database as DB
import user_crud
import schemas

from fastapi.middleware.cors import CORSMiddleware

origins = [
    "http://localhost:3000",
    "http://127.0.0.1:3000"
]

app = FastAPI()
app.add_middleware(
    CORSMiddleware,
    allow_origins=origins,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

DB.Base.metadata.create_all(bind=DB.engine)

def get_db():
    db = DB.SessionLocal()
    try:
        yield db
    finally:
        db.close()

user_crud.seedUsers(DB.SessionLocal())

@app.get('/users', response_model=List[schemas.UserResponseModel])
def getUsers(databaseInstance: Session = Depends(get_db)):
    users = user_crud.getAllUsers(databaseInstance)
    return users 

First, we import the required files (models, schemas, database, user_crud) and packages (fastapi, sqlalchemy.orm, typing).

We initiate our FastAPI app instance and then using the Base class and the engine which is in our database.py file we migrate the tables by using Base.metadata.create_all(). The metadata in the Base class is an instance of SQLAlchemy Metadata class.

We need to put in cors configs for security thus we are using fastapi's CORSMiddleware class. Then we will add those configs to our app using app.add_middleware()

We have a get_db() in our main.py. This function will help to create a Database session instance that we could use to connect and close the connection just before we return the response. Not just that, we need to have an independent database connection per request, use it through all the request and then close it. Thus we create a get_db dependency function with a yield to yield that independent database connection instance.

One thing we should carefully consider is the seedUsers(). We want to remove this function once the seeding is done. 

We have a route to get the list of users. We are expecting that each user in that list follows the UserResponseModel schema. So let's create that schema in schemas.py file.

from pydantic import BaseModel, EmailStr
from typing import List

class UserResponseModel(BaseModel):
    id: int
    name: str
    email: EmailStr
    class Config():
        orm_mode = True

Now let's start our server by running the following command

> uvicorn main:app --reload

Now in the browser, we need to open the URL http://localhost:8000/docs which should show something like below

When we run the user API we should get a list of 40 users.

Now the next part which is the main part of this article is pagination.

To add pagination we need to modify the getAllUsers function in user_crud.py We can use the skip and limit functions of SQLAlchemy but before that, we need to get the total count of users so that we know the last page number.

Also, we need to add UserDataModel pydantic model in our schemas.py. Along with that, we need to modify the getUsers function in our main.py so that we can pass in the page number as a query parameter and our response_model will now be of type UserDataModel.

So let's modify our code, step by step.

First schemas.py

from pydantic import BaseModel, EmailStr
from typing import List

class UserResponseModel(BaseModel):
    id: int
    name: str
    email: EmailStr
    class Config():
        orm_mode = True

class UserDataModel(BaseModel):
    data: List[UserResponseModel]
    total: int
    page: int
    per_page: int
    total_pages: int

Next main.py

from fastapi import Depends, FastAPI
from sqlalchemy.orm import Session
from typing import List
import models
import database as DB
import user_crud
import schemas

app = FastAPI()

DB.Base.metadata.create_all(bind=DB.engine)

def get_db():
    db = DB.SessionLocal()
    try:
        yield db
    finally:
        db.close()

user_crud.seedUsers(DB.SessionLocal())

@app.get('/users', response_model=schemas.UserDataModel)
def getUsers(page:int=1, databaseInstance: Session = Depends(get_db)):
    users = user_crud.getAllUsers(page, databaseInstance)
    return users 

And finally our user_crud.py file

import models
from faker import Faker

def getAllUsers(page, db):
    count = db.query(models.User).count()
    skip = (page - 1) * 10
    total_pages = count / limit if count % limit == 0 else count // limit + 1
    return {"data": db.query(models.User).offset(skip).limit(limit).all(), "total": count, "page": page, "per_page": limit, "total_pages": total_pages} 

def createUser(user, db):
    new_user = models.User(name=user.name, email=user.email)
    db.add(new_user)
    db.commit()
    db.refresh(new_user)
    return new_user


def seedUsers(db):
    fake = Faker()
    for i in range(40):
        user = models.User(name=fake.name(), email=fake.email())
        db.add(user)
        db.commit()

Time to test it by opening http://localhost:8000/docs

We can cache the results using caching but that's for some other day.

Well, that's all for today. Hope you enjoy reading this article. If you would like to learn to code along with developing your life skills then do check out our courses here.  Happy coding :)