How to package PAGE data？

[wu-clan]

Is it possible to not force the use of response_model as a Page class?

If I want to unify the return style of all interfaces, fastapi-pagination looks abrupt

Suppose my unified return model is

class ResponseModel(BaseModel): 
    code: int = 200 
    msg: str = 'Success' 
    data: Any | None = None

How to load page data under data parameters?

[uriyyo]

Hi @wu-clan,

Sorry, I do not fully understand what you want to achieve. Could you show an example?

[wu-clan]

E.g.:

This is a page api, Its return mode is Page[GetUserInfo]

@user.get('', summary='get all users')
async def get_all_users(db: Annotated[AsyncSession, Depends(get_db)]) -> Page[GetUserInfo]:
    users = select(User).order_by(User.time_joined.desc()).options(selectinload(User.roles).joinedload(Role.menus))
    return await paginate(db, users)

{
  "items": [...]
  ...
}

If I want to unify the return model as ResponseModel

@router.get('/{username}', summary='get userinfo by username')
async def get_userinfo(db: Annotated[AsyncSession, Depends(get_db)], username: str) -> ReasponseModel
    userinfo = select(User).where(username=username).options(selectinload(User.roles).joinedload(Role.menus))
    return ResponseModel(data=userinfo)

{
  "code": 200,
  "msg": "Success"
  "data": {
    ...
  }
}

So how should the paging interface be set up?

[uriyyo]

Hi @wu-clan,

I will recommend you convert ResponseModel into a generic model:

from typing import Generic, TypeVar

from pydantic import BaseModel
from pydantic.generics import GenericModel

DataT = TypeVar("DataT")


class ResponseModel(GenericModel, Generic[DataT]):
    code: int = 200
    msg: str = "Success"
    data: DataT | None = None

Then you will need to pagination_ctx to get_all_users route declaration and change response_model:

from fastapi_pagination import Page, pagination_ctx


@user.get(
    '',
    summary='get all users',
    dependencies=[Depends(pagination_ctx(Page))],
)
async def get_all_users(db: Annotated[AsyncSession, Depends(get_db)]) -> ResponseModel[list[UserOut]]:
    users = select(User).order_by(User.time_joined.desc()).options(selectinload(User.roles).joinedload(Role.menus))
    page = await paginate(db, users)

    return ResponseModel(data=page.items)

Or

from fastapi_pagination import Page, pagination_ctx


@user.get(
    '',
    summary='get all users',
    dependencies=[Depends(pagination_ctx(Page))],
)
async def get_all_users(db: Annotated[AsyncSession, Depends(get_db)]) -> ResponseModel[Page[UserOut]]:
    users = select(User).order_by(User.time_joined.desc()).options(selectinload(User.roles).joinedload(Role.menus))
    page = await paginate(db, users)

    return ResponseModel(data=page)

Please check the example bellow:

from datetime import datetime
from typing import Annotated, Any, AsyncIterator, Generic, TypeVar

from fastapi import Depends, FastAPI
from fastapi_pagination import Page, pagination_ctx
from fastapi_pagination.ext.sqlalchemy import paginate
from pydantic import BaseModel
from pydantic.generics import GenericModel
from sqlalchemy import select
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column

DataT = TypeVar("DataT")


class ResponseModel(GenericModel, Generic[DataT]):
    code: int = 200
    msg: str = "Success"
    data: DataT | None = None


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = "users"

    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column()
    email: Mapped[str] = mapped_column()
    created_at: Mapped[datetime] = mapped_column()


class UserOut(BaseModel):
    id: int
    name: str
    email: str
    created_at: datetime

    class Config:
        orm_mode = True


app = FastAPI()
engine = create_async_engine("postgresql+asyncpg://postgres:postgres@localhost:5433/postgres")


@app.on_event("startup")
async def on_startup() -> None:
    async with engine.connect() as conn:
        await conn.run_sync(Base.metadata.drop_all)
        await conn.run_sync(Base.metadata.create_all)
        await conn.commit()

    async with AsyncSession(engine) as session:
        session.add(User(name="user1", email="[email protected]", created_at=datetime.now()))
        await session.commit()


async def get_db() -> AsyncIterator[AsyncSession]:
    async with AsyncSession(engine) as session:
        yield session


def unwrap_page(page: Any) -> list[Any]:
    return page.items


@app.get(
    "/users",
    dependencies=[Depends(pagination_ctx(Page))],
)
async def get_all_users(db: Annotated[AsyncSession, Depends(get_db)]) -> ResponseModel[list[UserOut]]:
    query = select(User)
    page = await paginate(db, query)

    return ResponseModel(data=unwrap_page(page))

[wu-clan]

Hi, @uriyyo

I've fine-tuned it to fit my global code and it looks like

DataT = TypeVar('DataT')
SchemaT = TypeVar('SchemaT')

class _PageData(GenericModel, Generic[DataT]):
    page_data: DataT | None = None

async def paging_data(db: AsyncSession, select: Select, page_data_schema: SchemaT) -> dict:
    _paginate = await paginate(db, select)
    page_data = _PageData[Page[page_data_schema]](page_data=_paginate).dict()['page_data']
    return page_data

Then use Depends(pagination_ctx(Page)) and my fixed ResponseModel, it looks like

@app.get(
    "/users",
    dependencies=[Depends(pagination_ctx(Page))],
)
async def get_all_users(db: Annotated[AsyncSession, Depends(get_db)]) -> ResponseModel:
    select = select(User)
    page_data = paging_data(db, select, UserOut)
    return ResponseModel(data=page_data)

It works very well, thanks.

[uriyyo]

Great, happy to hear that ✨