Comments — One-to-Many with Ownership and Moderation

Comments are the canonical one-to-many relationship in web applications: one post has many comments, each comment belongs to exactly one post. Adding threaded replies introduces a self-referential one-to-many: one comment can have many reply comments, each pointing back to a parent via parent_comment_id. This lesson builds the complete comments API — create, list (with threading), moderate, and delete — showing how to chain related ownership checks (a post author can moderate any comment on their post, a commenter can delete their own comment) and how to load threaded comment trees efficiently.

Comment Model #

# app/models/comment.py
from sqlalchemy import Text, Boolean, ForeignKey, Integer
from sqlalchemy.orm import Mapped, mapped_column, relationship
from sqlalchemy.sql import func
from datetime import datetime
from app.database import Base

class Comment(Base):
    __tablename__ = "comments"

    id:               Mapped[int]            = mapped_column(primary_key=True)
    body:             Mapped[str]            = mapped_column(Text, nullable=False)
    is_approved:      Mapped[bool]           = mapped_column(Boolean, default=True)
    post_id:          Mapped[int]            = mapped_column(
                          ForeignKey("posts.id", ondelete="CASCADE"), index=True)
    author_id:        Mapped[int]            = mapped_column(
                          ForeignKey("users.id", ondelete="CASCADE"), index=True)
    parent_id:        Mapped[int | None]     = mapped_column(
                          ForeignKey("comments.id", ondelete="CASCADE"), nullable=True, index=True)
    created_at:       Mapped[datetime]       = mapped_column(server_default=func.now())
    updated_at:       Mapped[datetime]       = mapped_column(
                          server_default=func.now(), onupdate=func.now())

    # ── Relationships ─────────────────────────────────────────────────────────
    post:    Mapped["Post"]           = relationship(back_populates="comments")
    author:  Mapped["User"]           = relationship(back_populates="comments")
    # Self-referential: one comment can have many replies
    replies: Mapped[list["Comment"]]  = relationship(
        back_populates = "parent",
        cascade        = "all, delete-orphan",
    )
    parent:  Mapped["Comment | None"] = relationship(
        back_populates = "replies",
        remote_side    = [id],   # remote_side marks the "one" side of the self-join
    )

Comment Schemas #

# app/schemas/comment.py
from pydantic import BaseModel, Field
from datetime import datetime

class AuthorSummary(BaseModel):
    model_config = {"from_attributes": True}
    id:   int
    name: str

class CommentCreate(BaseModel):
    body:      str = Field(..., min_length=1, max_length=10000)
    parent_id: int | None = None   # None for top-level, int for reply

class CommentUpdate(BaseModel):
    body: str = Field(..., min_length=1, max_length=10000)

class CommentResponse(BaseModel):
    model_config = {"from_attributes": True}
    id:          int
    body:        str
    is_approved: bool
    post_id:     int
    parent_id:   int | None
    created_at:  datetime
    author:      AuthorSummary
    replies:     list["CommentResponse"] = []   # nested replies

CommentResponse.model_rebuild()   # needed for forward references in nested schema

Comments Router #

# app/routers/comments.py
from fastapi import APIRouter, Depends, HTTPException, status
from sqlalchemy import select
from sqlalchemy.orm import Session, joinedload, selectinload
from app.dependencies import get_db, get_current_user
from app.models.comment import Comment
from app.models.post    import Post
from app.schemas.comment import CommentCreate, CommentUpdate, CommentResponse

router = APIRouter()

@router.get("/posts/{post_id}/comments", response_model=list[CommentResponse])
def list_comments(post_id: int, db: Session = Depends(get_db)):
    """Return top-level approved comments with their replies."""
    comments = db.scalars(
        select(Comment)
        .options(
            joinedload(Comment.author),
            selectinload(Comment.replies).joinedload(Comment.author),
        )
        .where(
            Comment.post_id   == post_id,
            Comment.parent_id.is_(None),   # top-level only
            Comment.is_approved == True,
        )
        .order_by(Comment.created_at.asc())
    ).all()
    return comments

@router.post("/posts/{post_id}/comments",
             response_model=CommentResponse, status_code=201)
def create_comment(
    post_id:      int,
    comment_in:   CommentCreate,
    db:           Session = Depends(get_db),
    current_user = Depends(get_current_user),
):
    post = db.get(Post, post_id)
    if not post or post.deleted_at:
        raise HTTPException(404, "Post not found")

    # Validate parent comment belongs to same post
    if comment_in.parent_id:
        parent = db.get(Comment, comment_in.parent_id)
        if not parent or parent.post_id != post_id:
            raise HTTPException(400, "Invalid parent comment")
        # Limit nesting depth to 2 levels
        if parent.parent_id is not None:
            raise HTTPException(400, "Maximum reply depth exceeded")

    comment = Comment(
        body      = comment_in.body,
        post_id   = post_id,
        author_id = current_user.id,
        parent_id = comment_in.parent_id,
    )
    db.add(comment)
    db.flush()
    db.refresh(comment)
    return comment

@router.delete("/comments/{comment_id}", status_code=204)
def delete_comment(
    comment_id:   int,
    db:           Session = Depends(get_db),
    current_user = Depends(get_current_user),
):
    comment = db.get(Comment, comment_id)
    if not comment:
        raise HTTPException(404, "Comment not found")
    # Comment author OR post author can delete
    post = db.get(Post, comment.post_id)
    if comment.author_id != current_user.id and \
       (post and post.author_id != current_user.id) and \
       current_user.role != "admin":
        raise HTTPException(403, "Not authorised")
    db.delete(comment)
    db.flush()

Common Mistakes #

Mistake 1 — Missing remote_side on self-referential relationship #

❌ Wrong — SQLAlchemy cannot determine join direction:

parent: Mapped["Comment | None"] = relationship(back_populates="replies")
# AmbiguousForeignKeysError or wrong join!

✅ Correct — always specify remote_side for self-referential:

parent: Mapped["Comment | None"] = relationship(
    back_populates="replies", remote_side=[id]   # ✓
)

Mistake 2 — Not validating parent comment belongs to the same post #

❌ Wrong — client can attach a reply to a comment from a different post:

comment = Comment(body=body, post_id=post_id, parent_id=parent_id)
# parent_id might reference a comment on a completely different post!

✅ Correct — verify parent.post_id == post_id before creating the reply.

Mistake 3 — Loading all comments without the top-level filter #

❌ Wrong — returns both root comments and replies at the top level (duplicates replies):

select(Comment).where(Comment.post_id == post_id)
# Returns root comments AND replies — replies appear twice (as their own rows AND nested)

✅ Correct — filter for top-level only, load replies via selectinload:

select(Comment).where(Comment.post_id == post_id,
                      Comment.parent_id.is_(None))   # ✓

Quick Reference #