한양과학기술고등학교

에듀테크 프로젝트 대시보드

CRUD 라우트 구현

schemas from_attributes pagination PATCH 409

스키마 3종(Create/Update/Read)과 게시글 CRUD 전체 예시 (동기 SQLAlchemy 기준).

1. Pydantic 스키마

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

class PostCreate(BaseModel):
    title: str
    body: str = ""
    author_id: int

class PostUpdate(BaseModel):
    title: str | None = None
    body: str | None = None

class PostRead(BaseModel):
    model_config = ConfigDict(from_attributes=True)   # ORM -> 응답
    id: int
    title: str
    body: str
    author_id: int
    created_at: datetime

2. Create

from sqlalchemy import select
from sqlalchemy.exc import IntegrityError

@router.post("", response_model=PostRead, status_code=201)
def create_post(payload: PostCreate, db: DbDep):
    post = Post(**payload.model_dump())
    db.add(post)
    try:
        db.commit()
    except IntegrityError:
        db.rollback()
        raise HTTPException(409, "제약 조건 위반 (존재하지 않는 author_id 등)")
    db.refresh(post)
    return post

3. Read (목록 + 단건)

@router.get("", response_model=list[PostRead])
def list_posts(db: DbDep, skip: int = 0, limit: int = 20, q: str | None = None):
    stmt = select(Post)
    if q:
        stmt = stmt.where(Post.title.ilike(f"%{q}%"))
    stmt = stmt.order_by(Post.id.desc()).offset(skip).limit(min(limit, 100))
    return db.scalars(stmt).all()

@router.get("/{post_id}", response_model=PostRead)
def get_post(post_id: int, db: DbDep):
    post = db.get(Post, post_id)
    if post is None:
        raise HTTPException(404, "게시글 없음")
    return post

4. Update (부분 수정)

@router.patch("/{post_id}", response_model=PostRead)
def update_post(post_id: int, payload: PostUpdate, db: DbDep):
    post = db.get(Post, post_id)
    if post is None:
        raise HTTPException(404)
    data = payload.model_dump(exclude_unset=True)   # 보낸 필드만
    for key, value in data.items():
        setattr(post, key, value)
    db.commit()
    db.refresh(post)
    return post

exclude_unset=True 가 핵심 — 명시적으로 null 을 보낸 것과 아예 안 보낸 것을 구분.

5. Delete

@router.delete("/{post_id}", status_code=204)
def delete_post(post_id: int, db: DbDep):
    post = db.get(Post, post_id)
    if post is None:
        raise HTTPException(404)
    db.delete(post)
    db.commit()

6. 재사용 가능한 CRUD 헬퍼

def get_or_404(db: Session, model, pk):
    obj = db.get(model, pk)
    if obj is None:
        raise HTTPException(404, f"{model.__name__} not found")
    return obj
"타입 힌트 하나로 검증·직렬화·문서화가 동시에 이뤄진다"