Cursor Pagination
Another popular pagination type is cursor
.
It is similar to limit-offset
pagination type, but instead
of using offset
parameter, it uses cursor
parameter.
Cursor is a value that is used to identify the position of the last item in the previous page. It is usually a primary key of the last item in the previous page.
In this tutorial, you will learn how to use cursor
pagination type.
Note
cursor
pagination is only available for sqlalchemy
and casandra
backends.
Example
To use cursor
you need to import CursorPage
from fastapi_pagination.cursor
and use it as a response model.
from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
from fastapi_pagination import add_pagination
from fastapi_pagination.cursor import CursorPage
from fastapi_pagination.ext.sqlalchemy import paginate
app = FastAPI()
add_pagination(app)
engine = create_engine("sqlite:///.db")
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column()
age: Mapped[int] = mapped_column()
class UserOut(BaseModel):
id: int
name: str
age: int
class Config:
orm_mode = True
@app.on_event("startup")
def on_startup():
with engine.begin() as conn:
Base.metadata.drop_all(conn)
Base.metadata.create_all(conn)
with Session(engine) as session:
session.add_all(
[
User(name="John", age=25),
User(name="Jane", age=30),
User(name="Bob", age=20),
],
)
session.commit()
@app.get("/users")
def get_users() -> CursorPage[UserOut]:
with Session(engine) as session:
return paginate(session, select(User).order_by(User.id))
Total items behavior
By default CursorPage
doesn't include total count of items. If you want to include it, you should customize page
using UseIncludeTotal
customizer.
from typing import TypeVar
from fastapi import FastAPI
from pydantic import BaseModel
from sqlalchemy import create_engine, select
from sqlalchemy.orm import DeclarativeBase, Mapped, Session, mapped_column
from fastapi_pagination import add_pagination
from fastapi_pagination.cursor import CursorPage
from fastapi_pagination.customization import CustomizedPage, UseIncludeTotal
from fastapi_pagination.ext.sqlalchemy import paginate
app = FastAPI()
add_pagination(app)
engine = create_engine("sqlite:///.db")
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "users"
id: Mapped[int] = mapped_column(primary_key=True)
name: Mapped[str] = mapped_column()
age: Mapped[int] = mapped_column()
class UserOut(BaseModel):
id: int
name: str
age: int
class Config:
orm_mode = True
@app.on_event("startup")
def on_startup():
with engine.begin() as conn:
Base.metadata.drop_all(conn)
Base.metadata.create_all(conn)
with Session(engine) as session:
session.add_all(
[
User(name="John", age=25),
User(name="Jane", age=30),
User(name="Bob", age=20),
],
)
session.commit()
T = TypeVar("T")
CursorWithTotalPage = CustomizedPage[
CursorPage[T],
UseIncludeTotal(True),
]
@app.get("/users")
def get_users() -> CursorWithTotalPage[UserOut]:
with Session(engine) as session:
return paginate(session, select(User).order_by(User.id))