Pagination

Use Case 4: Pagination & Filtering

Scenario: Blog post API with pagination and filtering.

Python

from mongo_ops import BaseDocument, BaseRepository
from pydantic import BaseModel, Field
from typing import List, Optional, Generic, TypeVar

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: Optional[dict] = None,
        sort_by: str = "created_at",
        sort_order: int = -1
    ) -> PaginatedResponse[BlogPost]:
        """Get paginated blog posts"""
        filter_dict = filter_dict or {}
        skip = (page - 1) * page_size

        # Get total count
        total = await self.count(filter_dict)

        # Get items
        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):
        """Get posts by author"""
        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]:
        """Search posts by tags"""
        return await self.get_many(filter={"tags": {"$in": tags}, "published": True})

# Usage in FastAPI
@app.get("/posts/", response_model=PaginatedResponse[BlogPost])
async def list_posts(
    page: int = 1,
    page_size: int = 10,
    published: Optional[bool] = None,
    author_id: Optional[str] = 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)