Putting It Together — Complete Blog Post CRUD API

This lesson assembles all the patterns from the chapter into a complete, production-ready blog post CRUD router. The router handles all five HTTP methods, uses the service/repository architecture, implements soft delete, manages tags through the many-to-many relationship, returns consistent paginated responses, and applies proper authentication. This is the concrete, working implementation of the blog API that serves as the foundation for the rest of the series — the full-stack integration chapters will connect this API to a React frontend.

Complete Posts Router #

# app/routers/posts.py — complete implementation
from fastapi import APIRouter, Depends, HTTPException, Query, status
from sqlalchemy import select, func, update as sql_update, delete as sql_delete
from sqlalchemy.orm import Session, joinedload, selectinload
from sqlalchemy.exc import IntegrityError
from sqlalchemy.sql import func as sql_func
from typing import Annotated, Literal

from app.dependencies import get_db, get_current_user, require_auth
from app.models.post import Post
from app.models.tag  import Tag
from app.models.user import User
from app.schemas.post import PostCreate, PostUpdate, PostResponse, PostDetailResponse
from app.schemas.pagination import Page

router = APIRouter(
    prefix       = "/posts",
    tags         = ["Posts"],
    responses    = {404: {"description": "Post not found"}},
)

# ── LIST ──────────────────────────────────────────────────────────────────────
@router.get("/", response_model=Page[PostResponse])
def list_posts(
    page:      Annotated[int, Query(ge=1)] = 1,
    page_size: Annotated[int, Query(ge=1, le=100)] = 10,
    author_id: int | None = None,
    tag:       str | None = None,
    search:    str | None = None,
    sort_by:   Literal["created_at", "view_count", "title"] = "created_at",
    order:     Literal["asc", "desc"] = "desc",
    db:        Session = Depends(get_db),
):
    stmt = (
        select(Post)
        .options(joinedload(Post.author), selectinload(Post.tags))
        .where(Post.status == "published", Post.deleted_at.is_(None))
    )
    if author_id:
        stmt = stmt.where(Post.author_id == author_id)
    if tag:
        stmt = stmt.join(Post.tags).where(Tag.slug == tag).distinct()
    if search:
        stmt = stmt.where(
            Post.title.ilike(f"%{search}%") | Post.body.ilike(f"%{search}%")
        )
    total = db.scalar(select(func.count()).select_from(stmt.subquery()))
    sort_col = {"created_at": Post.created_at, "view_count": Post.view_count,
                "title": Post.title}[sort_by]
    ordered = sort_col.desc() if order == "desc" else sort_col.asc()
    posts = db.scalars(
        stmt.order_by(ordered, Post.id.desc())
            .offset((page - 1) * page_size).limit(page_size)
    ).all()
    return Page.create(items=posts, total=total, page=page, page_size=page_size)

# ── GET BY SLUG (before {post_id} to avoid route conflict) ────────────────────
@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

# ── GET BY ID ─────────────────────────────────────────────────────────────────
@router.get("/{post_id}", response_model=PostDetailResponse)
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(404, "Post not found")
    post.view_count += 1
    return post

# ── CREATE ────────────────────────────────────────────────────────────────────
@router.post("/", response_model=PostResponse, status_code=201,
             dependencies=[Depends(require_auth)])
def create_post(
    post_in:      PostCreate,
    db:           Session = Depends(get_db),
    current_user: User    = Depends(get_current_user),
):
    tag_objects = []
    if post_in.tags:
        for tag_name in post_in.tags:
            tag = db.scalars(select(Tag).where(Tag.name == tag_name)).first()
            if not tag:
                tag = Tag(name=tag_name, slug=tag_name.lower().replace(" ", "-"))
                db.add(tag)
                db.flush()
            tag_objects.append(tag)

    post_data = post_in.model_dump(exclude={"tags"})
    post = Post(**post_data, author_id=current_user.id, tags=tag_objects)
    db.add(post)
    try:
        db.flush()
        db.refresh(post)
        return post
    except IntegrityError:
        db.rollback()
        raise HTTPException(409, "A post with this slug already exists")

# ── PATCH ─────────────────────────────────────────────────────────────────────
@router.patch("/{post_id}", response_model=PostResponse,
              dependencies=[Depends(require_auth)])
def update_post(
    post_id:      int,
    update_in:    PostUpdate,
    db:           Session = Depends(get_db),
    current_user: User    = Depends(get_current_user),
):
    post = db.get(Post, post_id)
    if not post or post.deleted_at:
        raise HTTPException(404, "Post not found")
    if post.author_id != current_user.id and current_user.role != "admin":
        raise HTTPException(403, "Not your post")

    changes = update_in.model_dump(exclude_none=True, exclude={"tags"})
    for field, value in changes.items():
        setattr(post, field, value)

    # Update tags if provided
    if update_in.tags is not None:
        tag_objects = []
        for tag_name in update_in.tags:
            tag = db.scalars(select(Tag).where(Tag.name == tag_name)).first()
            if not tag:
                tag = Tag(name=tag_name, slug=tag_name.lower().replace(" ", "-"))
                db.add(tag); db.flush()
            tag_objects.append(tag)
        post.tags = tag_objects

    try:
        db.flush(); db.refresh(post)
        return post
    except IntegrityError:
        db.rollback()
        raise HTTPException(409, "Slug already in use")

# ── DELETE (soft) ─────────────────────────────────────────────────────────────
@router.delete("/{post_id}", status_code=204, dependencies=[Depends(require_auth)])
def delete_post(
    post_id:      int,
    db:           Session = Depends(get_db),
    current_user: User    = Depends(get_current_user),
):
    post = db.get(Post, post_id)
    if not post or post.deleted_at:
        raise HTTPException(404, "Post not found")
    if post.author_id != current_user.id and current_user.role != "admin":
        raise HTTPException(403, "Not your post")
    post.deleted_at = sql_func.now()
    db.flush()

Testing the Complete Router #

# tests/test_posts.py — basic smoke tests
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_list_posts_public():
    response = client.get("/posts/")
    assert response.status_code == 200
    data = response.json()
    assert "items" in data
    assert "total" in data

def test_create_post_unauthenticated():
    response = client.post("/posts/", json={"title": "Test", "body": "Content"})
    assert response.status_code == 401   # require_auth dependency fires

def test_get_nonexistent_post():
    response = client.get("/posts/999999")
    assert response.status_code == 404

Common Mistakes #

Mistake 1 — Route order: /{post_id} before /by-slug/{slug} #

❌ Wrong — /by-slug/{slug} never reached:

@router.get("/{post_id}")   # matches /by-slug/hello with post_id="by-slug"
@router.get("/by-slug/{slug}")   # never reached!

✅ Correct — specific routes first (as in the complete example above).

Mistake 2 — Not excluding tags from model_dump for post creation #

❌ Wrong — trying to pass tag strings to Post ORM model:

post = Post(**post_in.model_dump(), author_id=user.id)
# Post(**{"tags": ["python", "fastapi"]}) → TypeError! tags expects Tag objects

✅ Correct — exclude tags from model_dump, handle separately:

post = Post(**post_in.model_dump(exclude={"tags"}), author_id=user.id, tags=tag_objects)   # ✓

Quick Reference — Complete CRUD Pattern #