Read Endpoints — List with Pagination and Single Resource

GET endpoints are the majority of API traffic — every page load, every list refresh, every detail view hits a read endpoint. Well-designed read endpoints are fast (indexes, eager loading), predictable (consistent pagination envelope), and defensive (404 for missing resources, 403 for unauthorised access). This lesson builds the production-quality GET endpoints for the blog application — a paginated list with filtering, a detail endpoint with nested related data, and an “own posts” endpoint that returns only the authenticated user’s posts.

Paginated List Endpoint #

# app/routers/posts.py
from fastapi import APIRouter, Depends, Query, HTTPException, status
from sqlalchemy.orm import Session
from typing import Annotated, Literal
from app.dependencies import get_db
from app.schemas.post import PostResponse
from app.schemas.pagination import Page
from app.repositories import post_repo

router = APIRouter()

@router.get(
    "/",
    response_model     = Page[PostResponse],
    summary            = "List published posts",
    description        = "Returns a paginated list of published posts with optional filtering.",
)
def list_posts(
    page:      Annotated[int, Query(ge=1, description="Page number")] = 1,
    page_size: Annotated[int, Query(ge=1, le=100)] = 10,
    author_id: int | None = None,
    search:    Annotated[str | None, Query(max_length=200)] = None,
    sort_by:   Literal["created_at", "view_count", "title"] = "created_at",
    order:     Literal["asc", "desc"] = "desc",
    db:        Session = Depends(get_db),
):
    posts, total = post_repo.list_published_posts(
        db, page=page, page_size=page_size,
        author_id=author_id, search=search,
        sort_by=sort_by, order=order,
    )
    return Page.create(items=posts, total=total, page=page, page_size=page_size)

Single Resource with 404 Handling #

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session, joinedload, selectinload
from sqlalchemy import select
from app.models.post import Post
from app.schemas.post import PostDetailResponse

@router.get(
    "/{post_id}",
    response_model = PostDetailResponse,
    summary        = "Get a single post by ID",
)
def get_post(post_id: int, db: Session = Depends(get_db)):
    post = db.scalars(
        select(Post)
        .options(
            joinedload(Post.author),
            selectinload(Post.tags),
            selectinload(Post.comments).joinedload(Comment.author),
        )
        .where(Post.id == post_id, Post.deleted_at.is_(None))
    ).first()

    if not post:
        raise HTTPException(
            status_code = status.HTTP_404_NOT_FOUND,
            detail      = f"Post {post_id} not found",
        )

    # Increment view count (fire-and-forget — do not fail on error)
    try:
        post.view_count += 1
        db.flush()
    except Exception:
        pass   # view count increment is best-effort

    return post

# ── Get by slug (public URL) ──────────────────────────────────────────────────
@router.get("/by-slug/{slug}", response_model=PostDetailResponse)
def get_post_by_slug(slug: str, db: Session = Depends(get_db)):
    post = db.scalars(
        select(Post)
        .options(joinedload(Post.author), selectinload(Post.tags))
        .where(Post.slug == slug, Post.status == "published",
               Post.deleted_at.is_(None))
    ).first()
    if not post:
        raise HTTPException(404, "Post not found")
    return post

My Posts — Authenticated User’s Own Resources #

from app.dependencies import get_current_user
from app.models.user import User

@router.get(
    "/mine",
    response_model = Page[PostResponse],
    summary        = "List current user's posts (all statuses)",
)
def list_my_posts(
    page:      int = 1,
    page_size: int = 10,
    status:    str | None = None,  # filter by status
    db:        Session = Depends(get_db),
    current_user: User = Depends(get_current_user),
):
    stmt = (
        select(Post)
        .options(selectinload(Post.tags))
        .where(Post.author_id == current_user.id, Post.deleted_at.is_(None))
    )
    if status:
        stmt = stmt.where(Post.status == status)

    total = db.scalar(select(func.count()).select_from(stmt.subquery()))
    posts = db.scalars(
        stmt.order_by(Post.updated_at.desc())
            .offset((page - 1) * page_size)
            .limit(page_size)
    ).all()
    return Page.create(items=posts, total=total, page=page, page_size=page_size)

Common Mistakes #

Mistake 1 — Not filtering deleted_at in read queries #

❌ Wrong — soft-deleted posts appear in results:

posts = db.scalars(select(Post).where(Post.status == "published")).all()
# Returns deleted posts — soft delete is not applied!

✅ Correct — always filter deleted_at in every query:

posts = db.scalars(
    select(Post)
    .where(Post.status == "published", Post.deleted_at.is_(None))   # ✓
).all()

Mistake 2 — N+1 queries from lazy loading in list endpoint #

❌ Wrong — author loaded lazily for each post (N+1):

posts = db.scalars(select(Post).limit(20)).all()
return posts   # each post.author access → extra query!

✅ Correct — eager load with joinedload/selectinload.

Mistake 3 — No 404 on missing resource (returns None, serialisation fails) #

❌ Wrong — None returned, FastAPI cannot serialise it:

post = db.get(Post, post_id)
return post   # None → serialisation error or empty response

✅ Correct — always check and raise 404:

post = db.get(Post, post_id)
if not post:
    raise HTTPException(404, "Post not found")   # ✓
return post

Quick Reference #