Pydantic Pagination classes, Helps FastAPI to handle Pagination needed requests

api.py

from datetime import datetime
from typing import Generic, List, Optional, TypeVar

import factory
import uvicorn
from faker import Faker
from fastapi import FastAPI, Request
from pydantic import BaseModel, Field

fake = Faker()
T = TypeVar("T")


class Message(BaseModel):
    id: int
    text: str
    user_id: int
    chat_id: int
    created_at: datetime
    updated_at: datetime


class MessageFactory(factory.Factory):
    class Meta:
        model = Message

    id = factory.Sequence(lambda n: n)
    text = factory.Faker("sentence")
    user_id = factory.Faker("random_int", min=1, max=1000)
    chat_id = factory.Faker("random_int", min=1, max=100)
    created_at = factory.LazyFunction(datetime.utcnow)
    updated_at = factory.LazyFunction(datetime.utcnow)


class Pagination(BaseModel):
    offset: int = Field(1, ge=1)
    limit: int = Field(10, ge=1, le=100)


class PaginationResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    offset: int
    limit: int
    next_offset: Optional[int]
    prev_offset: Optional[int]

    @classmethod
    def create(
        cls, items: List[T], pagination: Pagination, total: int
    ) -> "PaginationResponse":
        return cls(
            items=items,
            total=total,
            offset=pagination.offset,
            limit=pagination.limit,
            next_offset=pagination.offset + pagination.limit
            if pagination.offset + pagination.limit < total
            else None,
            prev_offset=pagination.offset - pagination.limit
            if pagination.offset - pagination.limit > 0
            else None,
        )


def fake_data():
    return [MessageFactory() for _ in range(43)]


data = fake_data()


app = FastAPI()


@app.get("/")
def main(request: Request):
    return {"message": "Hello World"}


@app.get(
    "/get_messages",
)
async def main2(
    request: Request,
    chat_id: int,
    offset: int = 1,
    limit: int = 10,
) -> PaginationResponse[Message]:
    pagination = Pagination(offset=offset, limit=limit)
    items = data[pagination.offset - 1 : pagination.offset + pagination.limit - 1]

    return PaginationResponse.create(
        items=items,
        pagination=pagination,
        total=len(data),
    )


if __name__ == "__main__":
    uvicorn.run("api:app", host="0.0.0.0", port=7001, reload=True)

Pydantic-Pagination.py

from pydantic import BaseModel, Field
from typing import Optional, List, Generic, TypeVar

__all__ = ["Pagination", "PaginationResponse"]

T = TypeVar("T")

class Pagination(BaseModel):
    offset: int = Field(1, ge=1)
    limit: int = Field(10, ge=1, le=100)

class PaginationResponse(BaseModel, Generic[T]):
    items: List[T]
    total: int
    offset: int
    limit: int
    next_offset: Optional[int]
    prev_offset: Optional[int]

    @classmethod
    def create(
        cls, items: List[T], pagination: Pagination, total: int
    ) -> "PaginationResponse":
        return cls(
            items=items,
            total=total,
            offset=pagination.offset,
            limit=pagination.limit,
            next_offset=pagination.offset + pagination.limit
            if pagination.offset + pagination.limit < total
            else None,
            prev_offset=pagination.offset - pagination.limit
            if pagination.offset - pagination.limit > 0
            else None,
        )