Quartz 4

13. Pagination - Loading More Data with Query Parameters

15 min read

This tutorial implements pagination in the FastAPI application. The application now supports user registration, login, post creation, editing, deletion, and profile picture uploads. However, a significant problem exists with how posts are currently handled: all posts are returned at once. With only a handful of posts, this may not seem problematic. But with hundreds or thousands of posts, the application would become slow, wasteful, and provide a poor user experience.

Pagination fixes this issue by adding query parameters for skip and limit to the API, implementing database-level pagination with SQLAlchemy, and adding a “Load More” button on the frontend. This is an industry-standard pattern found in virtually every list endpoint in real-world APIs.

While pagination might seem like a frontend concern, it is actually driven by the backend API. The API controls how much data is sent per request, and the frontend simply consumes what the API provides. The majority of the implementation occurs on the API side, with frontend code wiring up to use it.

Generating Sample Data

Testing pagination requires sufficient data. A single post or even a handful makes testing difficult. The tutorial includes a utility script called populate_db.py that creates sample users, assigns profile pictures from a populate_images directory, and generates 44 sample posts with realistic dates spread over several months.

The script works by spinning up a test client and using the actual API endpoints rather than populating the database directly. It creates users, logs them in to get access tokens, uploads profile pictures, creates posts distributed across different users, and updates post dates to spread them over the past few months for realism.

Warning: This script clears all existing database content and replaces it with sample data. Ensure this is acceptable before running it. The script also requires profile images from the populate_images directory in the downloadable code—these can be swapped with custom images by updating the users list in the script.

Run the script:

uv run python populate_db.py

The script reports creating six users and 44 posts. Restart the development server and reload the homepage. All 44 posts now load on the page at once. With only 44 posts, this is already inefficient. Imagine the performance impact with thousands of posts—the application would send massive amounts of data over the network that users don’t need yet.

Defining the Paginated Response Schema

The first step is defining what a paginated response looks like. This establishes the contract between the API and frontend clients. Currently, the get_posts endpoint simply returns a list of posts. More information is needed—metadata about the pagination itself, such as how many total posts exist and whether more are available to load.

Open schemas.py and locate the PostResponse class. Add a new schema after it:

class PaginatedPostResponse(BaseModel):
    posts: list[PostResponse]
    total: int
    skip: int
    limit: int
    has_more: bool

The posts field contains the actual list of PostResponse objects—the post data itself. The total field is an integer representing the total count of posts in the database. The skip field represents the current offset (how many posts were skipped). The limit field indicates how many posts were requested. The has_more field is a boolean indicating whether more posts exist after this batch.

Why include has_more when total is already available? It simplifies frontend code. The frontend can check the has_more boolean and immediately know whether to show a “Load More” button without performing calculations. The API simply tells clients whether more posts exist, making client implementation easier.

Implementing API Pagination

Open routers/posts.py to implement pagination in the posts endpoint. Update the imports to include Query from FastAPI:

from fastapi import APIRouter, Depends, HTTPException, status, Query

Add func to the SQLAlchemy imports for count queries:

from sqlalchemy import select, func

Import the new paginated schema:

from schemas import PostCreate, PostResponse, PostUpdate, PaginatedPostResponse

Locate the get_posts endpoint and update it step by step.

Changing the Response Model

The current response model returns a list of PostResponse objects. Change it to return the new paginated schema:

@router.get("", response_model=PaginatedPostResponse)

Adding Query Parameters

Currently, the function only takes the database session parameter. Add skip and limit as query parameters:

async def get_posts(
    db: Annotated[AsyncSession, Depends(get_db)],
    skip: Annotated[int, Query(ge=0)] = 0,
    limit: Annotated[int, Query(ge=1, le=100)] = 10,
):

The Annotated syntax with Query adds constraints. For skip, ge=0 (greater than or equal to zero) prevents negative offsets, with a default of zero (start at the beginning if no skip is provided). For limit, ge=1 prevents requesting zero posts, le=100 caps requests at 100 to prevent someone from requesting millions of posts and exhausting resources, with a default of 10 posts per request.

Why use skip and limit instead of page and per_page? Skip and limit provide more flexibility. With skip and limit, clients can request any arbitrary range—skip=20, limit=10 returns posts 21 through 30. While page=3, per_page=10 is essentially equivalent, skip and limit are more common in REST APIs and give clients slightly more control.

Adding a Count Query

Before fetching posts, the total count is needed to calculate whether more posts exist. Add a count query before the existing select:

# Get total count
count_result = await db.execute(select(func.count()).select_from(models.Post))  
total = count_result.scalar() or 0

This queries the database using select(func.count()) to get the total number of posts. Order by is unnecessary since only a count is needed. If no result exists, default to zero.

Adding Offset and Limit to the Query

The existing query needs only minor modifications—add offset() and limit() methods:

# Get paginated posts
result = await db.execute(
    select(models.Post)
    .options(selectinload(models.Post.author))
    .order_by(models.Post.date_posted.desc())
    .offset(skip)
    .limit(limit)
)
posts = result.scalars().all()

The offset() method tells the database to skip that many rows before returning results. The limit() method tells it to return at most that many rows.

The order_by() is critical for pagination. Without ordering, the database can return rows in any order, meaning the same skip and limit values could produce different results on different requests. Ordering by date_posted in descending order ensures the newest posts always appear first, keeping pagination consistent.

Calculating has_more

After retrieving posts, calculate whether more exist. The logic: if skip plus the number of posts retrieved is less than total, more posts remain:

has_more = skip + len(posts) < total

For example, if 20 posts were skipped and 10 were retrieved, 30 posts have been seen. If the total is 44, more posts exist.

Returning the Paginated Response

Instead of returning just the list of posts, return the paginated response with all metadata:

return PaginatedPostResponse(
    posts=[PostResponse.model_validate(post) for post in posts],
    total=total,
    skip=skip,
    limit=limit,
    has_more=has_more,
)

Note the list comprehension: PostResponse.model_validate(post) for post in posts. Normally, FastAPI handles response model conversion automatically. But when constructing the response object manually, conversion must be handled explicitly. This ensures nested relationships like author are properly serialized.

Testing API Pagination

Navigate to the API documentation at /docs and locate the GET /api/posts endpoint. Expand it. The endpoint now displays skip and limit query parameters:

Click “Try it out” and execute with default values (skip=0, limit=10). The response returns a 200 success with the new paginated structure:

The response includes the first 10 posts (verify by counting). The total is 44, skip is 0 (default), limit is 10 (default), and has_more is true because more posts exist.

Test the next page with skip=10, limit=10. Execute the request. Different posts appear (the next 10), and the metadata shows total=44, skip=10, limit=10, has_more=true.

Test exhausting all posts with skip=40, limit=10. Execute the request. Only 4 posts return (44 total - 40 skipped = 4 remaining). The metadata shows total=44, skip=40, limit=10, has_more=false:

Testing Validation

Attempt to set limit=200 (exceeding the maximum of 100). The form prevents submission with a validation error: “Value must be less than or equal to 100”. The query constraints work correctly, preventing resource exhaustion:

The curl command can be copied from the documentation to test from the terminal if desired.

API-side pagination is complete. The frontend needs updating to use it.

Centralizing the Posts Per Page Setting

Before updating routes, centralize the posts-per-page setting using the same pattern as max_upload_size_bytes from the previous tutorial. Open config.py and add:

class Settings(BaseSettings):
    model_config = SettingsConfigDict(
        env_file=".env",
        env_file_encoding="utf-8"
    )
    
    secret_key: SecretStr
    algorithm: str = "HS256"
    access_token_expire_minutes: int = 30
    max_upload_size_bytes: int = 5 * 1024 * 1024
    posts_per_page: int = 10  # New setting

This allows changing the page size in one place if needed later.

Updating the Homepage Route

Open main.py. The homepage route currently performs its own database query directly. Even though the API endpoint at /api/posts is now paginated, the homepage template route still fetches every post from the database and renders them all.

Update the imports. Add func to SQLAlchemy imports:

from sqlalchemy import select, func

Import settings:

from config import settings

Locate the home route and replace it entirely:

@app.get("/", include_in_schema=False, name="home")
@app.get("/posts", include_in_schema=False, name="posts")
async def home(request: Request, db: Annotated[AsyncSession, Depends(get_db)]):
    count_result = await db.execute(select(func.count()).select_from(models.Post))
    total = count_result.scalar() or 0
 
    result = await db.execute(
        select(models.Post)
        .options(selectinload(models.Post.author))
        .order_by(models.Post.date_posted.desc())
        .limit(settings.posts_per_page),
    )
    posts = result.scalars().all()
 
    has_more = len(posts) < total
 
    return templates.TemplateResponse(
        request,
        "home.html",
        {
            "posts": posts,
            "title": "Home",
            "limit": settings.posts_per_page,
            "has_more": has_more,
        },
    )

This uses the same pagination logic as the API route. It gets the total count, fetches the first batch using posts_per_page from settings (no offset since this is always the first page), calculates has_more (no skip value to add since this is the initial load), and passes has_more to the template so JavaScript knows whether to show the “Load More” button.

This creates a hybrid approach: The first batch is server-side rendered for fast page loads and search engine visibility. Subsequent batches are fetched dynamically by JavaScript. This provides both fast initial loads and dynamic loading afterward.

Adding JavaScript Utilities

When rendering posts on the server with Jinja2, Jinja automatically escapes dangerous characters in user content. When injecting content via JavaScript, escaping must be done manually. Open static/js/utils.js and add two new utility functions:

// XSS prevention for dynamic content insertion  
export function escapeHtml(text) {  
  const div = document.createElement("div");  
  div.textContent = text;  
  return div.innerHTML;  
}  
  
// Date formatting to match server's strftime("%B %d, %Y")  
export function formatDate(dateString) {  
  const date = new Date(dateString);  
  return date.toLocaleDateString("en-US", {  
    year: "numeric",  
    month: "long",  
    day: "2-digit",  
  });  
}

The escapeHTML() function prevents cross-site scripting (XSS) attacks where someone puts malicious JavaScript in post titles or content that would execute in users’ browsers. It works by setting text using textContent (which treats everything as plain text), then reading it back with innerHTML (which provides the escaped version).

The formatDate() function converts ISO date strings (e.g., "2024-01-15T10:30:00Z") to match the format Jinja2 produces on the server side (e.g., "1/15/2024").

Both functions are exported for use in templates.

Updating the Home Template

Open templates/home.html. Two small changes are needed to the content block.

Wrap the for loop in a container div so JavaScript can append new posts:

{% block content %}  
    <div id="postsContainer">  
        {% for post in posts %}  
        ...
        {% endfor %}  
    </div>  
    {% if has_more %}  
        <div class="text-center mb-4">  
            <button type="button" class="btn btn-outline-primary" id="loadMoreBtn">Load More Posts</button>  
        </div>    {% endif %}  
{% endblock content %}

The “Load More” button only renders if has_more is true. If no more posts exist, it doesn’t appear at all.

Add a scripts block at the end of the file:

{% block scripts %}  
    <script type="module">  
  import { escapeHtml, formatDate } from '/static/js/utils.js';  
  
  // Pagination state - initialized from server-rendered values  
  let currentOffset = {{ limit }};  // Start after server-rendered posts  
  const limit = {{ limit }};  
  let hasMore = {{ 'true' if has_more else 'false' }};  
  
  const postsContainer = document.getElementById('postsContainer');  
  const loadMoreBtn = document.getElementById('loadMoreBtn');  
  
  // Create HTML for a single post (matching server-rendered structure)  
  function createPostHTML(post) {  
    return `  
      <article class="content-section py-3 px-4 mb-4">        <div class="d-flex align-items-start gap-4">          <img class="rounded-circle article-img flex-shrink-0" src="${escapeHtml(post.author.image_path)}" alt="${escapeHtml(post.author.username)}'s profile picture" width="64" height="64" loading="lazy">  
          <div class="flex-grow-1">            <div class="article-metadata mb-2">              <a class="me-2" href="/users/${post.author.id}/posts">${escapeHtml(post.author.username)}</a>  
              <small class="text-body-secondary">${formatDate(post.date_posted)}</small>  
            </div>            <h2>              <a class="article-title" href="/posts/${post.id}">${escapeHtml(post.title)}</a>  
            </h2>            <p class="article-content">${escapeHtml(post.content)}</p>  
          </div>        </div>      </article>    `;  
  }  
  
  // Load more posts from the API  
  async function loadMorePosts() {  
    // Disable button and show loading state  
    loadMoreBtn.disabled = true;  
    loadMoreBtn.textContent = 'Loading...';  
  
    let errorOccurred = false;  
  
    try {  
      const response = await fetch(`/api/posts?skip=${currentOffset}&limit=${limit}`);  
  
      if (!response.ok) {  
        throw new Error('Failed to fetch posts');  
      }  
  
      const data = await response.json();  
  
      // Append new posts to the container  
      for (const post of data.posts) {  
        postsContainer.insertAdjacentHTML('beforeend', createPostHTML(post));  
      }  
  
      // Update pagination state  
      currentOffset += data.posts.length;  
      hasMore = data.has_more;  
  
      // Hide button if no more posts  
      if (!hasMore) {  
        loadMoreBtn.classList.add('d-none');  
      }  
    } catch (error) {  
      errorOccurred = true;  
      console.error('Error loading posts:', error);  
      // Show error message and keep button enabled for retry  
      loadMoreBtn.textContent = 'Error - Click to Retry';  
      loadMoreBtn.disabled = false;  
    } finally {  
      // Re-enable button and reset text only if no error occurred  
      if (!errorOccurred && hasMore) {  
        loadMoreBtn.disabled = false;  
        loadMoreBtn.textContent = 'Load More Posts';  
      }  
    }  
  }  
  
  // Add click handler if button exists  
  if (loadMoreBtn) {  
    loadMoreBtn.addEventListener('click', loadMorePosts);  
  }  
    </script>  
{% endblock scripts %}

The script initializes pagination state using Jinja2 to inject server-rendered values. currentOffset starts at limit because the server already rendered the first batch—fetching should begin from there. hasMore determines whether the button should even exist.

The createPostHTML() function builds HTML for a single post. This must match the server-rendered post structure exactly. It uses escapeHTML() for all user content and formatDate() for dates.

The loadMorePosts() function contains the main logic. It fetches from the paginated API using the current offset and limit, handles errors, appends new posts to the container, updates the current offset using the number of posts returned, and hides the “Load More” button if no more posts exist. If an error occurs, the button changes to “Error - Click to retry” so users can try again.

Testing Frontend Pagination

Reload the homepage. The first 10 posts render (verify by counting). A “Load More Posts” button appears at the bottom.

Click the button. The application dynamically fetches the next 10 posts from the API and appends them to the page. The button remains visible because more posts still exist.

Continue clicking until all 44 posts load. When no more posts exist, the button disappears.

Implementing Pagination for User Posts

The same pattern should apply to user-specific post pages. Navigate to a user’s posts page (click on a username). These pages currently lack pagination.

Updating the User Posts API Endpoint

Open routers/users.py and add Query to the FastAPI imports:

from fastapi import APIRouter, Depends, HTTPException, status, Query

Add PaginatedPostResponse to the schema imports:

from schemas import PostResponse, UserCreate, UserPrivate, UserPublic, UserUpdate, Token, PaginatedPostResponse

Locate the get_user_posts endpoint and replace it:

@router.get("/{user_id}/posts", response_model=PaginatedPostsResponse)  
async def get_user_posts(  
    user_id: int,  
    db: Annotated[AsyncSession, Depends(get_db)],  
    skip: Annotated[int, Query(ge=0)] = 0,  
    limit: Annotated[int, Query(ge=1, le=100)] = settings.posts_per_page,  
):  
    result = await db.execute(select(models.User).where(models.User.id == user_id))  
    user = result.scalars().first()  
    if not user:  
        raise HTTPException(  
            status_code=status.HTTP_404_NOT_FOUND,  
            detail="User not found",  
        )  
  
    count_result = await db.execute(  
        select(func.count())  
        .select_from(models.Post)  
        .where(models.Post.user_id == user_id),  
    )  
    total = count_result.scalar() or 0  
  
    result = await db.execute(  
        select(models.Post)  
        .options(selectinload(models.Post.author))  
        .where(models.Post.user_id == user_id)  
        .order_by(models.Post.date_posted.desc())  
        .offset(skip)  
        .limit(limit),  
    )  
    posts = result.scalars().all()  
  
    has_more = skip + len(posts) < total  
  
    return PaginatedPostsResponse(  
        posts=[PostResponse.model_validate(post) for post in posts],  
        total=total,  
        skip=skip,  
        limit=limit,  
        has_more=has_more,  
    )

This follows exactly the same pattern as the main posts endpoint. The only difference is filtering by user_id in both the count query and the select query.

Updating the User Posts Template Route

Open main.py and locate the user_posts_page route. Replace it with:

@app.get("/users/{user_id}/posts", include_in_schema=False, name="user_posts")
async def user_posts_page(
    request: Request,
    user_id: int,
    db: Annotated[AsyncSession, Depends(get_db)],
):
    result = await db.execute(select(models.User).where(models.User.id == user_id))
    user = result.scalars().first()
    if not user:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail="User not found",
        )
 
    count_result = await db.execute(
        select(func.count())
        .select_from(models.Post)
        .where(models.Post.user_id == user_id),
    )
    total = count_result.scalar() or 0
 
    result = await db.execute(
        select(models.Post)
        .options(selectinload(models.Post.author))
        .where(models.Post.user_id == user_id)
        .order_by(models.Post.date_posted.desc())
        .limit(settings.posts_per_page),
    )
    posts = result.scalars().all()
 
    has_more = len(posts) < total
 
    return templates.TemplateResponse(
        request,
        "user_posts.html",
        {
            "posts": posts,
            "user": user,
            "title": f"{user.username}'s Posts",
            "limit": settings.posts_per_page,
            "has_more": has_more,
        },
    )

Same pattern: get the count filtered by user ID, fetch the first batch, calculate has_more, and pass limit and has_more to the template.

Updating the User Posts Template

The user_posts.html template is almost identical to home.html with two differences: a heading shows whose posts are being viewed, and JavaScript fetches from a different API URL that includes the user ID (look at the repo for the code).

The structure is essentially identical to home.html. The key difference is the fetch URL: /api/users/${userId}/posts instead of /api/posts.

Testing User Posts Pagination

Navigate to a user’s posts page by clicking a username. If the user has fewer than 10 posts, no “Load More” button appears. The sample data likely doesn’t include any users with more than 10 posts, making full testing difficult. However, the API endpoint can be tested via the documentation.

Navigate to /docs and test GET /api/users/{user_id}/posts with user_id=1. If the response shows fewer than 10 posts total, has_more correctly returns false.

Alternative: FastAPI Pagination Library

A library called fastapi-pagination handles much of this boilerplate automatically. It supports multiple pagination strategies and provides consistent response formats. For learning purposes, manual implementation is better for understanding what happens under the hood. For production applications, evaluate whether the library saves time versus adding complexity—that decision depends on project needs and team preferences.

Summary

This tutorial added complete pagination to the FastAPI application. Skip and limit query parameters were added to API endpoints with validation (preventing negative offsets, limiting maximum requests to 100). A new PaginatedPostResponse schema was created to provide metadata alongside post data. SQLAlchemy’s offset() and limit() methods implement database-level pagination. Frontend “Load More” buttons fetch additional pages from the API dynamically.

The result is a real pagination system where the backend controls data flow and the frontend consumes it. The first page loads via server-side rendering for speed and SEO, while subsequent pages load dynamically via JavaScript.

The next tutorial covers password reset functionality, including sending emails with background tasks, creating secure reset tokens, and completing the password reset placeholder currently on the account page.

Graph View