Skip to content

Use Case 4: Pagination & Filtering

Scenario: A blog post API lists posts with page metadata, filtering, and sorting.

📦 What's New?

Component Description
BaseRepository.get_many filter, skip, limit, sort in one call.
BaseRepository.count Total matching documents for page metadata.
PaginatedResponse Reusable generic response wrapper.

🚀 Example

Python
from typing import Generic, TypeVar

from fastapi import FastAPI, Query
from pydantic import BaseModel
from mongo_ops import BaseDocument, BaseRepository

T = TypeVar("T")


class PaginatedResponse(BaseModel, Generic[T]):
    items: list[T]
    total: int
    page: int
    page_size: int
    total_pages: int
    has_next: bool
    has_prev: bool


class BlogPost(BaseDocument):
    title: str = ""
    content: str = ""
    author_id: str = ""
    published: bool = False
    tags: list[str] = []
    views: int = 0


class BlogPostRepository(BaseRepository[BlogPost]):
    def __init__(self):
        super().__init__("blog_posts", BlogPost)

    async def paginate(
        self,
        page: int = 1,
        page_size: int = 10,
        filter_dict: dict | None = None,
        sort_by: str = "created_at",
        sort_order: int = -1,
    ) -> PaginatedResponse[BlogPost]:
        filter_dict = filter_dict or {}
        skip = (page - 1) * page_size
        total = await self.count(filter_dict)
        items = await self.get_many(
            filter=filter_dict,
            skip=skip,
            limit=page_size,
            sort=[(sort_by, sort_order)],
        )
        total_pages = (total + page_size - 1) // page_size
        return PaginatedResponse(
            items=items,
            total=total,
            page=page,
            page_size=page_size,
            total_pages=total_pages,
            has_next=page < total_pages,
            has_prev=page > 1,
        )

    async def get_by_author(self, author_id: str, published_only: bool = True) -> list[BlogPost]:
        filter_dict = {"author_id": author_id}
        if published_only:
            filter_dict["published"] = True
        return await self.get_many(filter=filter_dict, sort=[("created_at", -1)])

    async def search_by_tags(self, tags: list[str]) -> list[BlogPost]:
        return await self.get_many(filter={"tags": {"$in": tags}, "published": True})


app = FastAPI()
blog_repo = BlogPostRepository()


@app.get("/posts/", response_model=PaginatedResponse[BlogPost])
async def list_posts(
    page: int = 1,
    page_size: int = 10,
    published: bool | None = None,
    author_id: str | None = None,
):
    filter_dict = {}
    if published is not None:
        filter_dict["published"] = published
    if author_id:
        filter_dict["author_id"] = author_id
    return await blog_repo.paginate(page, page_size, filter_dict)


@app.get("/posts/author/{author_id}", response_model=list[BlogPost])
async def posts_by_author(author_id: str, published: bool = True):
    return await blog_repo.get_by_author(author_id, published)


@app.get("/posts/tags", response_model=list[BlogPost])
async def posts_by_tags(tags: list[str] = Query(...)):
    return await blog_repo.search_by_tags(tags)

Note: add the lifespan wiring from use case 01 so blog_repo is created after connection.

💡 Tips

  • get_many calls cursor.to_list(limit); pass limit=0 to skip the limit entirely, otherwise a large explicit limit is safer than unbounded reads.
  • Always count first for stable metadata — it uses the same filter as the page query.
  • Combine with sort=[(field, -1|1)] for deterministic ordering; create matching indexes to avoid full collection scans (see use case 13).