Write Endpoints — Create, Update and Partial Update

Write endpoints — POST to create, PUT to fully replace, PATCH to partially update — require more defensive programming than read endpoints. Each write must validate ownership (you can only edit your own posts), handle database constraint violations gracefully (duplicate slug → 409), and return the correct status code (201 for create, 200 for update). Using separate PostCreate, PostUpdate, and PostResponse schemas for each operation (as established in Chapter 23) keeps each endpoint focused and prevents clients from setting fields they should not.

Create Endpoint (POST) #

from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy.orm import Session
from sqlalchemy.exc import IntegrityError
from app.schemas.post import PostCreate, PostResponse
from app.models.post import Post

router = APIRouter()

@router.post(
    "/",
    response_model = PostResponse,
    status_code    = status.HTTP_201_CREATED,
    summary        = "Create a new blog post",
)
def create_post(
    post_in:      PostCreate,
    db:           Session = Depends(get_db),
    current_user: User    = Depends(get_current_user),
):
    try:
        post = Post(
            **post_in.model_dump(),
            author_id = current_user.id,
        )
        db.add(post)
        db.flush()
        db.refresh(post)
        return post
    except IntegrityError as e:
        db.rollback()
        if "posts_slug_key" in str(e.orig) or "unique" in str(e.orig).lower():
            raise HTTPException(
                status_code = status.HTTP_409_CONFLICT,
                detail      = "A post with this slug already exists",
            )
        raise HTTPException(status.HTTP_400_BAD_REQUEST, "Database error")

Full Update Endpoint (PUT) #

from app.schemas.post import PostCreate, PostResponse

# PUT replaces the entire resource — client sends all fields
@router.put(
    "/{post_id}",
    response_model = PostResponse,
    summary        = "Replace a post (full update)",
)
def replace_post(
    post_id:      int,
    post_in:      PostCreate,   # full PostCreate — all required fields
    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 authorised to edit this post")

    try:
        for field, value in post_in.model_dump().items():
            setattr(post, field, value)
        db.flush()
        db.refresh(post)
        return post
    except IntegrityError:
        db.rollback()
        raise HTTPException(409, "Slug already in use by another post")

Partial Update Endpoint (PATCH) #

from app.schemas.post import PostUpdate

# PATCH updates only the provided fields — client sends subset
@router.patch(
    "/{post_id}",
    response_model = PostResponse,
    summary        = "Partially update a post",
)
def update_post(
    post_id:      int,
    update_in:    PostUpdate,   # all fields optional
    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 authorised")

    # Only update fields that were explicitly provided
    changes = update_in.model_dump(exclude_none=True)
    if not changes:
        return post   # nothing to update — return unchanged

    try:
        for field, value in changes.items():
            setattr(post, field, value)
        db.flush()
        db.refresh(post)
        return post
    except IntegrityError:
        db.rollback()
        raise HTTPException(409, "Slug already in use")

Common Mistakes #

Mistake 1 — Trusting author_id from request body #

❌ Wrong — client can set any author:

post = Post(**post_in.model_dump())   # model_dump() includes any author_id the client sent!

✅ Correct — always override with authenticated user:

post = Post(**post_in.model_dump(), author_id=current_user.id)   # ✓

Mistake 2 — PATCH using model_dump() without exclude_none #

❌ Wrong — sets unmentioned fields to None:

for field, value in update_in.model_dump().items():
    setattr(post, field, value)   # body=None if client only sent title!

✅ Correct:

for field, value in update_in.model_dump(exclude_none=True).items():
    setattr(post, field, value)   # ✓ only sent fields updated

Mistake 3 — No ownership check on update (IDOR vulnerability) #

❌ Wrong — any user can update any post:

post = db.get(Post, post_id)
if not post: raise HTTPException(404)
# No check: who owns this post?
for field, value in update.items(): setattr(post, field, value)

✅ Correct — check ownership before modifying:

if post.author_id != current_user.id and current_user.role != "admin":
    raise HTTPException(403, "Not your post")   # ✓

Quick Reference #