IT 20년차의 복합 비즈니스 도전

안녕하세요! 이번 글에서는 지난 장들에서 배운 내용을 종합하여 Poetry 기반 FastAPI 백엔드를 실제로 간단한 블로그 API 형태로 구현해보겠습니다. 데이터베이스 연동부터 인증(로그인/권한), 비동기 작업, 테스트, 그리고 Docker를 통한 배포까지 한 번에 살펴보며, 실제 운영 환경에 가까운 시나리오를 체험해 볼 수 있습니다.

목표

• User 모델과 게시글(Post), 댓글(Comment) 기능을 구현
• JWT 인증을 통한 로그인/회원가입
• CRUD API 작성과 테스트
• Docker Compose로 한꺼번에 배포 & 실행

---

8.1. 프로젝트 요구사항 정의

8.1.1. 기능 목록

1. 회원가입 & 로그인
   • email + password로 회원가입
   • 비밀번호는 해시(bcrypt) 적용
   • 로그인 시 JWT 토큰 발급
2. 게시글(Post) CRUD
   • 제목, 내용, 작성자(유저)
   • 게시글 수정/삭제는 작성자 본인만 가능 (권한)
3. 댓글(Comment) CRUD
   • 게시글에 달리는 댓글
   • 작성자 본인만 삭제 가능
4. 목록 조회, 상세 조회
   • 페이지네이션은 최소화된 형태로 예시 구현
5. 관리자(Admin)
   • Admin Role(역할)을 가진 유저는 모든 게시글·댓글을 삭제 가능
6. 비동기 예시
   • 예: 새 게시글 등록 시 알림 이메일 전송 (Celery로 구현, 선택 기능)

8.1.2. 기술 스택

• Python 3.10+
• FastAPI
• SQLAlchemy + Alembic
• JWT (python-jose, passlib)
• Celery + Redis (선택)
• Docker & Docker Compose
• Pytest (테스트)

---

8.2. 프로젝트 구조 설계

이전 장에서 언급한 디렉토리 구조 예시를 재확인해 봅시다.

app/
├── main.py               # FastAPI 진입점
├── core/                 # 설정, 보안, 유틸
│   └── config.py
├── db/
│   ├── session.py        # 세션/엔진
│   ├── base.py           # Base 모델
│   └── migrations/       # Alembic 마이그레이션
├── models/
│   ├── user.py
│   ├── post.py
│   └── comment.py
├── schemas/
│   ├── user.py
│   ├── post.py
│   └── comment.py
├── crud/
│   ├── user.py
│   ├── post.py
│   └── comment.py
├── api/
│   └── v1/
│       ├── endpoints/
│       │   ├── auth.py
│       │   ├── user.py
│       │   ├── post.py
│       │   └── comment.py
│       └── routers.py
├── celery_app.py         # Celery 초기화
├── tasks.py              # 비동기 작업 예시
└── tests/
    └── ...               # 테스트 파일들

---

8.3. 모델 정의

8.3.1. User 모델

# app/models/user.py
from sqlalchemy import Column, Integer, String
from sqlalchemy.orm import relationship
from app.db.base import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String, unique=True, index=True, nullable=False)
    hashed_password = Column(String, nullable=False)
    role = Column(String, default="user")  # "admin" or "user"

    posts = relationship("Post", back_populates="user")
    comments = relationship("Comment", back_populates="user")

8.3.2. Post 모델

# app/models/post.py
from sqlalchemy import Column, Integer, String, Text, ForeignKey
from sqlalchemy.orm import relationship
from app.db.base import Base

class Post(Base):
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String, nullable=False)
    content = Column(Text, nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)

    user = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post", cascade="all, delete-orphan")

8.3.3. Comment 모델

# app/models/comment.py
from sqlalchemy import Column, Integer, Text, ForeignKey
from sqlalchemy.orm import relationship
from app.db.base import Base

class Comment(Base):
    __tablename__ = "comments"

    id = Column(Integer, primary_key=True, index=True)
    content = Column(Text, nullable=False)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    post_id = Column(Integer, ForeignKey("posts.id"), nullable=False)

    user = relationship("User", back_populates="comments")
    post = relationship("Post", back_populates="comments")

---

8.4. 스키마 & CRUD 레이어

8.4.1. Pydantic 스키마 (예: schemas/post.py)

# app/schemas/post.py
from pydantic import BaseModel
from typing import Optional, List

class PostBase(BaseModel):
    title: str
    content: str

class PostCreate(PostBase):
    pass

class PostUpdate(BaseModel):
    title: Optional[str] = None
    content: Optional[str] = None

class PostRead(PostBase):
    id: int
    user_id: int

    class Config:
        orm_mode = True

class PostReadWithComments(PostRead):
    comments: List["CommentRead"] = []

from app.schemas.comment import CommentRead
PostReadWithComments.update_forward_refs()

• PostReadWithComments: 게시글 조회 시 댓글 목록까지 반환하고 싶다면, 관계된 스키마를 포함
• 순환 참조(Cyclic import) 문제를 피하기 위해 update_forward_refs()를 활용

8.4.2. CRUD 예시 (예: crud/post.py)

# app/crud/post.py
from sqlalchemy.orm import Session
from app.models.post import Post
from app.schemas.post import PostCreate, PostUpdate

def create_post(db: Session, user_id: int, post_in: PostCreate) -> Post:
    db_post = Post(
        title=post_in.title,
        content=post_in.content,
        user_id=user_id
    )
    db.add(db_post)
    db.commit()
    db.refresh(db_post)
    return db_post

def get_post(db: Session, post_id: int) -> Post:
    return db.query(Post).filter(Post.id == post_id).first()

def update_post(db: Session, db_post: Post, post_in: PostUpdate) -> Post:
    if post_in.title is not None:
        db_post.title = post_in.title
    if post_in.content is not None:
        db_post.content = post_in.content
    db.commit()
    db.refresh(db_post)
    return db_post

def delete_post(db: Session, db_post: Post):
    db.delete(db_post)
    db.commit()

def get_posts(db: Session, skip: int = 0, limit: int = 10):
    return db.query(Post).offset(skip).limit(limit).all()

---

8.5. 인증 & 권한 로직

8.5.1. Auth 엔드포인트 (api/v1/endpoints/auth.py)

# app/api/v1/endpoints/auth.py
from fastapi import APIRouter, Depends, HTTPException
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session

from app.api.deps import get_db
from app.core.security import create_access_token, verify_password
from app.crud.user import get_user_by_email, create_user
from app.schemas.token import Token

router = APIRouter()

@router.post("/login", response_model=Token)
def login(db: Session = Depends(get_db), form_data: OAuth2PasswordRequestForm = Depends()):
    user = get_user_by_email(db, form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=400, detail="Incorrect email or password")
    access_token = create_access_token({"sub": str(user.id)})
    return {"access_token": access_token, "token_type": "bearer"}

@router.post("/signup", response_model=Token)
def signup(email: str, password: str, db: Session = Depends(get_db)):
    existing_user = get_user_by_email(db, email)
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    new_user = create_user(db, email, password)
    # 가입 즉시 로그인 토큰 발급
    access_token = create_access_token({"sub": str(new_user.id)})
    return {"access_token": access_token, "token_type": "bearer"}

8.5.2. 권한 확인 (api/deps.py)

# app/api/deps.py
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer
from jose import JWTError, jwt
from sqlalchemy.orm import Session

from app.db.session import SessionLocal
from app.models.user import User
from app.crud.user import get_user_by_id
from app.core.config import settings
from app.core.security import SECRET_KEY, ALGORITHM

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="login")

def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

def get_current_user(token: str = Depends(oauth2_scheme), db: Session = Depends(get_db)) -> User:
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: str = payload.get("sub")
    except JWTError:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="Invalid token")
    if not user_id:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="No user ID in token")
    user = get_user_by_id(db, int(user_id))
    if not user:
        raise HTTPException(status_code=status.HTTP_401_UNAUTHORIZED, detail="User not found")
    return user

def get_admin_user(current_user: User = Depends(get_current_user)):
    if current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Not enough permissions")
    return current_user

---

8.6. 주요 엔드포인트: Post & Comment

8.6.1. Post 엔드포인트 (api/v1/endpoints/post.py)

from fastapi import APIRouter, Depends, HTTPException
from typing import List
from sqlalchemy.orm import Session

from app.api.deps import get_db, get_current_user, get_admin_user
from app.crud.post import create_post, get_post, get_posts, update_post, delete_post
from app.models.user import User
from app.schemas.post import PostCreate, PostRead, PostUpdate

router = APIRouter()

@router.get("/", response_model=List[PostRead])
def read_posts(
    skip: int = 0,
    limit: int = 10,
    db: Session = Depends(get_db)
):
    return get_posts(db, skip, limit)

@router.get("/{post_id}", response_model=PostRead)
def read_post(post_id: int, db: Session = Depends(get_db)):
    db_post = get_post(db, post_id)
    if not db_post:
        raise HTTPException(status_code=404, detail="Post not found")
    return db_post

@router.post("/", response_model=PostRead, status_code=201)
def create_new_post(
    post_in: PostCreate,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    return create_post(db, current_user.id, post_in)

@router.put("/{post_id}", response_model=PostRead)
def update_existing_post(
    post_id: int,
    post_in: PostUpdate,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    db_post = get_post(db, post_id)
    if not db_post:
        raise HTTPException(status_code=404, detail="Post not found")
    # 작성자거나 Admin이 아니면 수정 불가
    if db_post.user_id != current_user.id and current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Not enough permissions")
    return update_post(db, db_post, post_in)

@router.delete("/{post_id}")
def delete_existing_post(
    post_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    db_post = get_post(db, post_id)
    if not db_post:
        raise HTTPException(status_code=404, detail="Post not found")
    # 작성자거나 Admin이 아니면 삭제 불가
    if db_post.user_id != current_user.id and current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Not enough permissions")
    delete_post(db, db_post)
    return {"detail": "Post deleted"}

8.6.2. Comment 엔드포인트

Comment 엔드포인트는 Post와 매우 유사합니다. 주의할 점은 게시글 ID가 필요하다는 정도입니다.

# app/api/v1/endpoints/comment.py
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.api.deps import get_db, get_current_user
from app.crud.comment import create_comment, get_comment, delete_comment
from app.schemas.comment import CommentCreate, CommentRead
from app.models.user import User

router = APIRouter()

@router.post("/", response_model=CommentRead, status_code=201)
def create_new_comment(
    post_id: int,
    comment_in: CommentCreate,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    return create_comment(db, post_id, current_user.id, comment_in)

@router.delete("/{comment_id}")
def delete_existing_comment(
    comment_id: int,
    current_user: User = Depends(get_current_user),
    db: Session = Depends(get_db)
):
    db_comment = get_comment(db, comment_id)
    if not db_comment:
        raise HTTPException(status_code=404, detail="Comment not found")
    # 작성자나 admin 권한 확인
    if db_comment.user_id != current_user.id and current_user.role != "admin":
        raise HTTPException(status_code=403, detail="Not enough permissions")
    delete_comment(db, db_comment)
    return {"detail": "Comment deleted"}

---

8.7. Celery 비동기 작업 (선택 기능)

# app/tasks.py
from app.celery_app import celery_app
import time

@celery_app.task
def send_notification_email(post_id: int):
    # 실제로는 이메일 서버 연동
    time.sleep(5)  # 예: 이메일 전송에 걸리는 시간
    print(f"[Celery] Notification email sent for Post ID {post_id}")

• 게시글 생성 시 이 작업을 호출:

  # create_new_post 함수 내부에서
  # send_notification_email.delay(new_post.id)
  

---

8.8. Docker Compose로 실행

8.8.1. Dockerfile

FROM python:3.10-slim

WORKDIR /app

RUN pip install --upgrade pip && pip install poetry

COPY pyproject.toml poetry.lock /app/
RUN poetry install --no-dev --no-interaction --no-ansi

COPY . /app

CMD ["poetry", "run", "uvicorn", "app.main:app", "--host", "0.0.0.0", "--port", "80"]

8.8.2. docker-compose.yml

version: '3.8'
services:
  web:
    build: .
    container_name: fastapi-blog
    ports:
      - "80:80"
    depends_on:
      - redis
    # 환경 변수 예시
    environment:
      - DB_HOST=db
      - DB_USER=postgres
      - DB_PASSWORD=secret
      - REDIS_URL=redis://redis:6379/0

  worker:
    build: .
    container_name: celery-worker
    command: poetry run celery -A app.celery_app.celery_app worker --loglevel=info
    depends_on:
      - redis

  db:
    image: postgres:14
    container_name: postgres-db
    environment:
      POSTGRES_USER: postgres
      POSTGRES_PASSWORD: secret
    ports:
      - "5432:5432"

  redis:
    image: redis:6.2
    container_name: redis
    ports:
      - "6379:6379"

8.8.3. 실행 및 마이그레이션

# 빌드 & 실행
docker-compose up --build

# 컨테이너에 들어가 alembic 마이그레이션 적용 (또는 별도 셸 스크립트)
docker-compose exec web poetry run alembic upgrade head

---

8.9. 테스트 & 시연

8.9.1. 간단한 테스트 시나리오

1. 회원가입: POST /api/v1/auth/signup?email=test@example.com&password=123456
2. 로그인: POST /api/v1/auth/login (OAuth2PasswordRequestForm) → JWT 토큰 획득
3. 게시글 작성: POST /api/v1/posts 헤더에 Authorization: Bearer <token>

   {
     "title": "Hello Blog",
     "content": "This is my first post"
   }
   

4. 게시글 조회: GET /api/v1/posts
5. 댓글 작성: POST /api/v1/comments?post_id=1 (JWT 필수)

   {
     "content": "Nice post!"
   }
   

6. 게시글 삭제: 작성자 본인 또는 admin 유저의 토큰이 있어야 가능

8.9.2. TestClient를 이용한 자동화 테스트

# tests/test_blog.py
from fastapi.testclient import TestClient
from app.main import app

client = TestClient(app)

def test_signup_and_login():
    signup_resp = client.post("/api/v1/auth/signup?email=test2@example.com&password=123456")
    assert signup_resp.status_code == 200
    data = signup_resp.json()
    assert "access_token" in data

    token = data["access_token"]
    headers = {"Authorization": f"Bearer {token}"}

    # Create post
    post_resp = client.post("/api/v1/posts", headers=headers, json={
        "title": "Test Post",
        "content": "Test Content"
    })
    assert post_resp.status_code == 201
    post_data = post_resp.json()
    assert post_data["title"] == "Test Post"
    assert post_data["content"] == "Test Content"

---

8.10. 마무리

이번 8장에서는 간단한 블로그 API를 예시로 해서, 우리가 이전 장에서 학습한 FastAPI + SQLAlchemy + JWT 인증 + (선택적으로 Celery, Docker) 등 핵심 기능들을 실제 프로젝트에 녹여보았습니다. 주요 포인트를 다시 짚어 보면:

1. DB 모델링: User, Post, Comment 간의 관계 설정
2. Pydantic 스키마와 CRUD 레이어를 통해 깔끔한 아키텍처
3. JWT 인증 및 **권한(Role)**으로 API 접근 제어
4. 비동기 작업(Celery)을 통해 장시간 프로세스를 백그라운드에서 처리
5. Docker Compose로 전체 스택(web, worker, db, redis)을 일괄 관리 및 배포

이제 여기서 더 확장하면, 파일 업로드(이미지/동영상), 알림 시스템, 여러 환경(Dev/Staging/Prod) 세팅 등에 도전해 볼 수 있습니다. 또한, AWS/GCP/Kubernetes 같은 인프라 환경에 배포해 스케일링을 할 수도 있고요.