Skip to content

데이터베이스 연동

현대 웹 애플리케이션은 대부분 데이터베이스에 의존하여 데이터를 저장, 조회, 수정, 삭제하는 CRUD 작업을 수행합니다. FastAPI는 ORM(Object-Relational Mapping)을 직접 지원하지 않지만, 다양한 써드파티 ORM 라이브러리와 원활하게 연동할 수 있습니다.

주요 Python ORM 라이브러리

파이썬 생태계에는 다양한 ORM 옵션이 있으며, 각각의 특징과 장단점을 가지고 있습니다:

  • SQLAlchemy : ⭐ 10.5k - 가장 성숙하고 강력한 ORM
    • Alembic : ⭐ 3.4k - SQLAlchemy용 마이그레이션 도구
  • SQLModel : ⭐ 16.1k - FastAPI 창시자가 만든 현대적 ORM
  • Tortoise ORM : ⭐ 5.1k - Django ORM과 유사한 비동기 ORM
  • ormar : ⭐ 1.7k - FastAPI와 Pydantic 통합에 특화
  • Piccolo : ⭐ 1.6k - 현대적이고 직관적인 비동기 ORM
  • edgy : ⭐ 279 - 새로운 현대적 ORM

이 가이드에서는 가장 널리 사용되고 안정적인 SQLAlchemy를 중심으로 설명하겠습니다.

라이브러리 설치

기본적으로 SQLAlchemy만 설치하면 됩니다:

pip install sqlalchemy

데이터베이스 드라이버

SQLite는 파이썬에 기본 내장되어 있어 추가 설치가 불필요합니다.

다른 데이터베이스를 사용하는 경우 해당 드라이버를 설치해야 합니다:

  • PostgreSQL: pip install psycopg2-binary
  • MySQL: pip install PyMySQL
  • SQL Server: pip install pyodbc

데이터베이스 설정

기본 설정 파일

database.py
import os
from contextlib import contextmanager
from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker

# 환경변수에서 DATABASE_URL을 읽어오고, 없으면 기본값으로 SQLite 사용
SQLALCHEMY_DATABASE_URL = os.getenv("DATABASE_URL", "sqlite:///./db.sqlite3")

# 데이터베이스 엔진 생성
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, 
    connect_args={"check_same_thread": False}  # SQLite 전용: 멀티스레드 지원
)

# 세션 팩토리 생성
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 모델 클래스의 기본 클래스
Base = declarative_base()

def get_db_session():
    """
    FastAPI 의존성 주입용 데이터베이스 세션 생성

    엔드포인트 함수에서 Depends()와 함께 사용하여
    자동으로 세션을 생성하고 정리합니다.
    """
    db = SessionLocal()
    try:
        yield db
    finally:
        # 함수 실행 완료 또는 오류 발생 시 항상 연결 정리
        db.close()

@contextmanager
def get_db_context():
    """
    컨텍스트 매니저를 사용한 데이터베이스 세션 관리

    with 문과 함께 사용하여 자동으로 세션을 관리합니다.
    주로 스크립트나 배치 작업에서 사용합니다.
    """
    db = SessionLocal()
    try:
        yield db
        db.commit()  # 성공 시 자동 커밋
    except Exception:
        db.rollback()  # 오류 시 자동 롤백
        raise
    finally:
        db.close()

__all__ = ["Base", "get_db_session", "get_db_context"]

환경별 데이터베이스 설정

config.py
import os
from functools import lru_cache

class Settings:
    """애플리케이션 설정"""

    def __init__(self):
        self.database_url = self._get_database_url()
        self.debug = os.getenv("DEBUG", "False").lower() == "true"

    def _get_database_url(self) -> str:
        """환경에 따른 데이터베이스 URL 반환"""
        env = os.getenv("ENVIRONMENT", "development")

        if env == "production":
            return os.getenv("DATABASE_URL", "postgresql://user:pass@localhost/db")
        elif env == "testing":
            return "sqlite:///./test.db"
        else:  # development
            return "sqlite:///./dev.db"

@lru_cache()
def get_settings():
    """설정 객체 싱글톤 반환"""
    return Settings()

SQL을 통한 직접 데이터베이스 조회

Raw SQL 사용 예시

main.py
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy import text
from sqlalchemy.orm import Session
from database import get_db_session

app = FastAPI()

@app.get("/posts/")
async def post_list(db: Session = Depends(get_db_session)):
    """Raw SQL을 사용한 포스트 목록 조회"""

    query = text("""
        SELECT id, title, content, created_at 
        FROM posts 
        ORDER BY created_at DESC 
        LIMIT 10
    """)

    try:
        result = db.execute(query)
        posts = [dict(row._mapping) for row in result]
        return {"posts": posts}
    except Exception as e:
        raise HTTPException(status_code=500, detail=f"Database error: {str(e)}")

Raw SQL 사용 시점

다음과 같은 경우에 Raw SQL을 고려해보세요:

  • 복잡한 집계 쿼리나 윈도우 함수가 필요한 경우
  • 데이터베이스 특화 기능을 사용해야 하는 경우
  • 성능 최적화가 중요한 복잡한 쿼리
  • 기존 SQL 쿼리를 마이그레이션하는 경우

SQLAlchemy ORM 모델 정의

기본 모델 클래스

models.py
from datetime import datetime
from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean
from sqlalchemy.sql import func
from database import Base

class Post(Base):
    """블로그 포스트 모델"""
    __tablename__ = "posts"

    # 기본 키
    id = Column(Integer, primary_key=True, index=True)

    # 기본 필드
    title = Column(String(200), nullable=False, index=True)
    content = Column(Text, nullable=False)
    is_published = Column(Boolean, default=False, index=True)

    # 자동 생성 타임스탬프
    created_at = Column(
        DateTime(timezone=True), 
        server_default=func.now(),
        nullable=False
    )
    updated_at = Column(
        DateTime(timezone=True), 
        server_default=func.now(),
        onupdate=func.now(),
        nullable=False
    )

    def __repr__(self):
        return f"<Post(id={self.id}, title='{self.title[:30]}...')>"

    def __str__(self):
        return f"{self.title}"

class User(Base):
    """사용자 모델"""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    email = Column(String(255), unique=True, index=True, nullable=False)
    username = Column(String(50), unique=True, index=True, nullable=False)
    hashed_password = Column(String(255), nullable=False)
    is_active = Column(Boolean, default=True)

    created_at = Column(DateTime(timezone=True), server_default=func.now())

    def __repr__(self):
        return f"<User(id={self.id}, username='{self.username}')>"

고급 모델 기능

models_advanced.py
from sqlalchemy import Column, Integer, String, ForeignKey, Table
from sqlalchemy.orm import relationship
from database import Base

# 다대다 관계를 위한 연결 테이블
post_tags = Table(
    'post_tags',
    Base.metadata,
    Column('post_id', Integer, ForeignKey('posts.id'), primary_key=True),
    Column('tag_id', Integer, ForeignKey('tags.id'), primary_key=True)
)

class Post(Base):
    """포스트 모델 (관계 포함)"""
    __tablename__ = "posts"

    id = Column(Integer, primary_key=True, index=True)
    title = Column(String(200), nullable=False)
    content = Column(Text, nullable=False)

    # 외래 키
    author_id = Column(Integer, ForeignKey('users.id'), nullable=False)

    # 관계 정의
    author = relationship("User", back_populates="posts")
    comments = relationship("Comment", back_populates="post", cascade="all, delete-orphan")
    tags = relationship("Tag", secondary=post_tags, back_populates="posts")

class User(Base):
    """사용자 모델 (관계 포함)"""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(50), unique=True, nullable=False)

    # 관계 정의
    posts = relationship("Post", back_populates="author")
    comments = relationship("Comment", back_populates="author")

class Comment(Base):
    """댓글 모델"""
    __tablename__ = "comments"

    id = Column(Integer, primary_key=True, index=True)
    content = Column(Text, nullable=False)

    # 외래 키
    post_id = Column(Integer, ForeignKey('posts.id'), nullable=False)
    author_id = Column(Integer, ForeignKey('users.id'), nullable=False)

    # 관계 정의
    post = relationship("Post", back_populates="comments")
    author = relationship("User", back_populates="comments")

class Tag(Base):
    """태그 모델"""
    __tablename__ = "tags"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(50), unique=True, nullable=False)

    # 관계 정의
    posts = relationship("Post", secondary=post_tags, back_populates="tags")

테이블 생성 및 관리

CLI 도구를 통한 테이블 생성

cli.py
#!/usr/bin/env python3
"""
데이터베이스 관리를 위한 CLI 도구

사용법:
    python cli.py create-tables  # 테이블 생성
    python cli.py drop-tables    # 테이블 삭제 (주의!)
"""

import typer
from rich.console import Console
from rich.prompt import Confirm

from database import engine
from models import Base

app = typer.Typer(help="데이터베이스 관리 도구")
console = Console()

@app.command()
def create_tables():
    """
    모델 정의를 기반으로 데이터베이스 테이블을 생성합니다.

    주의사항:
    - 이미 존재하는 테이블은 변경되지 않습니다
    - 컬럼 추가/수정/삭제는 마이그레이션 도구(Alembic) 사용이 필요합니다
    - 개발 초기 단계나 새로운 환경 설정 시에만 사용하세요
    """
    try:
        Base.metadata.create_all(bind=engine)
        console.print("✅ 모든 테이블이 성공적으로 생성되었습니다.", style="green")
    except Exception as e:
        console.print(f"❌ 테이블 생성 중 오류 발생: {e}", style="red")
        raise typer.Exit(1)

@app.command()
def drop_tables():
    """
    모든 테이블을 삭제합니다. ⚠️ 주의: 모든 데이터가 삭제됩니다!
    """
    if not Confirm.ask("⚠️  모든 테이블과 데이터가 삭제됩니다. 계속하시겠습니까?"):
        console.print("작업이 취소되었습니다.", style="yellow")
        return

    try:
        Base.metadata.drop_all(bind=engine)
        console.print("✅ 모든 테이블이 삭제되었습니다.", style="green")
    except Exception as e:
        console.print(f"❌ 테이블 삭제 중 오류 발생: {e}", style="red")
        raise typer.Exit(1)

@app.command()
def init_db():
    """데이터베이스를 초기화합니다 (테이블 생성 + 초기 데이터)"""
    create_tables()
    # 여기에 초기 데이터 생성 로직 추가 가능

if __name__ == "__main__":
    app()

# 설치 및 사용
pip install typer rich

# 테이블 생성
python cli.py create-tables

# 도움말 확인
python cli.py --help

애플리케이션 시작 시 자동 초기화

main.py
from contextlib import asynccontextmanager
from fastapi import FastAPI

@asynccontextmanager
async def lifespan(app: FastAPI):
    """애플리케이션 라이프사이클 관리"""
    # 시작 시 실행
    from database import engine
    from models import Base

    # 개발 환경에서만 자동 테이블 생성
    import os
    if os.getenv("ENVIRONMENT") == "development":
        Base.metadata.create_all(bind=engine)
        print("✅ 개발 환경: 테이블 자동 생성 완료")

    yield

    # 종료 시 실행 (필요한 경우)
    print("🔄 애플리케이션 종료 중...")

app = FastAPI(lifespan=lifespan)

프로덕션 환경에서의 주의사항

Base.metadata.create_all()은 기존 테이블 구조를 변경하지 않습니다.

환경별 권장사항:

  • 개발 환경: 테이블 삭제 후 재생성 가능
  • 스테이징/프로덕션: Alembic 마이그레이션 도구 필수 사용
# Alembic 설치 및 초기 설정
pip install alembic
alembic init alembic
alembic revision --autogenerate -m "Initial migration"
alembic upgrade head

SQLAlchemy ORM을 활용한 데이터 조작

기본 CRUD 작업

crud_examples.py
from models import Post, User
from database import get_db_context

def basic_crud_examples():
    """기본적인 CRUD 작업 예시"""

    with get_db_context() as db:

        # 1. CREATE - 새 레코드 생성
        new_post = Post(
            title="FastAPI와 SQLAlchemy 활용하기",
            content="이 글에서는 FastAPI와 SQLAlchemy를 함께 사용하는 방법을 알아봅니다.",
            is_published=True
        )

        db.add(new_post)
        db.commit()
        db.refresh(new_post)  # DB에서 자동 생성된 값들(id, created_at 등) 동기화

        print(f"✅ 새 포스트 생성: {new_post.id} - {new_post.title}")
        created_post_id = new_post.id

        # 2. READ - 데이터 조회

        # 전체 레코드 조회 (페이징)
        posts = db.query(Post).offset(0).limit(10).all()
        print(f"📋 전체 포스트 수: {len(posts)}")

        # 특정 레코드 조회
        post = db.query(Post).filter(Post.id == created_post_id).first()
        print(f"🔍 조회된 포스트: {post.title}")

        # 조건부 조회
        published_posts = db.query(Post).filter(Post.is_published == True).all()
        print(f"📢 게시된 포스트 수: {len(published_posts)}")

        # 3. UPDATE - 데이터 수정
        post.title = "FastAPI와 SQLAlchemy 완전 정복"
        post.content += "\n\n업데이트된 내용입니다."

        # 이미 세션에 있는 객체는 add() 불필요
        db.commit()
        db.refresh(post)
        print(f"✏️  포스트 수정 완료: {post.title}")

        # 4. DELETE - 데이터 삭제
        db.delete(post)
        db.commit()
        print("🗑️  포스트 삭제 완료")

def advanced_query_examples():
    """고급 쿼리 예시"""

    with get_db_context() as db:

        # 정렬과 필터링
        recent_posts = (
            db.query(Post)
            .filter(Post.is_published == True)
            .order_by(Post.created_at.desc())
            .limit(5)
            .all()
        )

        # 부분 문자열 검색
        search_posts = (
            db.query(Post)
            .filter(Post.title.contains("FastAPI"))
            .all()
        )

        # 집계 함수
        from sqlalchemy import func

        post_count = db.query(func.count(Post.id)).scalar()
        latest_post_date = db.query(func.max(Post.created_at)).scalar()

        print(f"📊 통계: 총 {post_count}개 포스트, 최신 포스트: {latest_post_date}")

        # 관계를 통한 조인 쿼리 (관계가 정의된 경우)
        # posts_with_authors = (
        #     db.query(Post)
        #     .join(User)
        #     .filter(User.is_active == True)
        #     .all()
        # )

if __name__ == "__main__":
    basic_crud_examples()
    advanced_query_examples()

Pydantic 스키마 정의

요청/응답 스키마

schemas.py
from datetime import datetime
from typing import Optional, List
from pydantic import BaseModel, Field, validator

class PostBase(BaseModel):
    """포스트 기본 스키마"""
    title: str = Field(..., min_length=1, max_length=200, description="포스트 제목")
    content: str = Field(..., min_length=1, description="포스트 내용")
    is_published: bool = Field(default=False, description="게시 여부")

class PostCreate(PostBase):
    """포스트 생성 스키마"""

    @validator('title')
    def validate_title(cls, v):
        """제목 검증"""
        if not v.strip():
            raise ValueError('제목은 공백일 수 없습니다')
        return v.strip()

class PostUpdate(BaseModel):
    """포스트 수정 스키마 (부분 수정 지원)"""
    title: Optional[str] = Field(None, min_length=1, max_length=200)
    content: Optional[str] = Field(None, min_length=1)
    is_published: Optional[bool] = None

    @validator('title')
    def validate_title(cls, v):
        if v is not None and not v.strip():
            raise ValueError('제목은 공백일 수 없습니다')
        return v.strip() if v else v

class PostResponse(PostBase):
    """포스트 응답 스키마"""
    id: int
    created_at: datetime
    updated_at: datetime

    class Config:
        from_attributes = True  # SQLAlchemy 모델과 호환 (v2)
        # orm_mode = True  # Pydantic v1에서는 이 설정 사용

class PostListResponse(BaseModel):
    """포스트 목록 응답 스키마"""
    posts: List[PostResponse]
    total: int
    page: int
    per_page: int
    has_next: bool
    has_prev: bool

# 사용자 관련 스키마
class UserBase(BaseModel):
    """사용자 기본 스키마"""
    username: str = Field(..., min_length=3, max_length=50)
    email: str = Field(..., regex=r'^[^@]+@[^@]+\.[^@]+$')

class UserCreate(UserBase):
    """사용자 생성 스키마"""
    password: str = Field(..., min_length=8, description="최소 8자 이상")

class UserResponse(UserBase):
    """사용자 응답 스키마"""
    id: int
    is_active: bool
    created_at: datetime

    class Config:
        from_attributes = True

REST API 엔드포인트 구현

포스트 CRUD API

main.py
from typing import List, Optional
from fastapi import FastAPI, HTTPException, Depends, status, Query
from fastapi.responses import JSONResponse
from sqlalchemy.orm import Session
from sqlalchemy import func

# 로컬 모듈
from database import get_db_session
from models import Post
from schemas import PostCreate, PostUpdate, PostResponse, PostListResponse

app = FastAPI(
    title="Blog API",
    description="FastAPI와 SQLAlchemy를 활용한 블로그 API",
    version="1.0.0"
)

@app.get("/posts/", response_model=PostListResponse)
def post_list(
    page: int = Query(1, ge=1, description="페이지 번호"),
    per_page: int = Query(10, ge=1, le=100, description="페이지당 항목 수"),
    search: Optional[str] = Query(None, description="제목 검색"),
    published_only: bool = Query(True, description="게시된 포스트만 조회"),
    db: Session = Depends(get_db_session)
):
    """포스트 목록 조회 (페이징, 검색, 필터링 지원)"""

    # 기본 쿼리 구성
    query = db.query(Post)

    # 필터 적용
    if published_only:
        query = query.filter(Post.is_published == True)

    if search:
        query = query.filter(Post.title.contains(search))

    # 전체 개수 조회
    total = query.count()

    # 페이징 적용
    offset = (page - 1) * per_page
    posts = (
        query
        .order_by(Post.created_at.desc())
        .offset(offset)
        .limit(per_page)
        .all()
    )

    return PostListResponse(
        posts=posts,
        total=total,
        page=page,
        per_page=per_page,
        has_next=offset + per_page < total,
        has_prev=page > 1
    )

@app.get("/posts/{post_id}/", response_model=PostResponse)
def get_post(post_id: int, db: Session = Depends(get_db_session)):
    """포스트 상세 조회"""

    post = db.query(Post).filter(Post.id == post_id).first()

    if not post:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"포스트 ID {post_id}를 찾을 수 없습니다."
        )

    return post

@app.post("/posts/", response_model=PostResponse, status_code=status.HTTP_201_CREATED)
def post_new(
    post_data: PostCreate, 
    db: Session = Depends(get_db_session)
):
    """새 포스트 생성"""

    try:
        # 모델 인스턴스 생성
        db_post = Post(**post_data.dict())

        # 데이터베이스에 저장
        db.add(db_post)
        db.commit()
        db.refresh(db_post)

        return db_post

    except Exception as e:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"포스트 생성 중 오류가 발생했습니다: {str(e)}"
        )

@app.patch("/posts/{post_id}/", response_model=PostResponse)
def post_edit(
    post_id: int,
    post_update: PostUpdate,
    db: Session = Depends(get_db_session)
):
    """포스트 부분 수정"""

    # 기존 포스트 조회
    db_post = db.query(Post).filter(Post.id == post_id).first()

    if not db_post:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"포스트 ID {post_id}를 찾을 수 없습니다."
        )

    # 업데이트할 데이터만 추출 (None 값 제외)
    update_data = post_update.dict(exclude_unset=True)

    if not update_data:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="수정할 데이터가 없습니다."
        )

    try:
        # 필드별 업데이트
        for field, value in update_data.items():
            setattr(db_post, field, value)

        db.commit()
        db.refresh(db_post)

        return db_post

    except Exception as e:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"포스트 수정 중 오류가 발생했습니다: {str(e)}"
        )

@app.delete("/posts/{post_id}/", status_code=status.HTTP_204_NO_CONTENT)
def post_delete(post_id: int, db: Session = Depends(get_db_session)):
    """포스트 삭제"""

    db_post = db.query(Post).filter(Post.id == post_id).first()

    if not db_post:
        raise HTTPException(
            status_code=status.HTTP_404_NOT_FOUND,
            detail=f"포스트 ID {post_id}를 찾을 수 없습니다."
        )

    try:
        db.delete(db_post)
        db.commit()

    except Exception as e:
        db.rollback()
        raise HTTPException(
            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
            detail=f"포스트 삭제 중 오류가 발생했습니다: {str(e)}"
        )

@app.get("/posts/stats/", response_model=dict)
def get_post_stats(db: Session = Depends(get_db_session)):
    """포스트 통계 조회"""

    total_posts = db.query(func.count(Post.id)).scalar()
    published_posts = db.query(func.count(Post.id)).filter(Post.is_published == True).scalar()
    latest_post = db.query(func.max(Post.created_at)).scalar()

    return {
        "total_posts": total_posts,
        "published_posts": published_posts,
        "draft_posts": total_posts - published_posts,
        "latest_post_date": latest_post
    }

# 에러 핸들러
@app.exception_handler(404)
async def not_found_handler(request, exc):
    return JSONResponse(
        status_code=404,
        content={"detail": "요청한 리소스를 찾을 수 없습니다."}
    )

@app.exception_handler(500)
async def internal_error_handler(request, exc):
    return JSONResponse(
        status_code=500,
        content={"detail": "서버 내부 오류가 발생했습니다."}
    )

성능 최적화 및 베스트 프랙티스

연결 풀 설정

database_optimized.py
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.pool import QueuePool

# 프로덕션용 최적화된 엔진 설정
engine = create_engine(
    DATABASE_URL,
    poolclass=QueuePool,
    pool_size=20,                    # 기본 연결 수
    max_overflow=30,                 # 최대 추가 연결 수
    pool_pre_ping=True,              # 연결 상태 확인
    pool_recycle=3600,               # 1시간마다 연결 재생성
    echo=False,                      # SQL 로깅 (개발 시에만 True)
    connect_args={
        "check_same_thread": False,  # SQLite 전용
        "timeout": 30,               # 연결 타임아웃
    }
)

쿼리 최적화

optimized_queries.py
from sqlalchemy.orm import selectinload, joinedload
from models import Post, User, Comment

def optimized_query_examples(db: Session):
    """최적화된 쿼리 예시"""

    # N+1 문제 해결: 연관 데이터 미리 로딩
    posts_with_authors = (
        db.query(Post)
        .options(joinedload(Post.author))  # JOIN으로 한 번에 로딩
        .filter(Post.is_published == True)
        .all()
    )

    # 컬렉션 관계 최적화
    posts_with_comments = (
        db.query(Post)
        .options(selectinload(Post.comments))  # 별도 쿼리로 효율적 로딩
        .all()
    )

    # 필요한 컬럼만 선택
    from sqlalchemy import select

    post_titles = db.execute(
        select(Post.id, Post.title)
        .where(Post.is_published == True)
        .order_by(Post.created_at.desc())
        .limit(10)
    ).all()

    # 배치 처리를 위한 청크 단위 조회
    def process_posts_in_chunks(chunk_size=1000):
        offset = 0
        while True:
            posts = (
                db.query(Post)
                .offset(offset)
                .limit(chunk_size)
                .all()
            )

            if not posts:
                break

            # 포스트 처리
            for post in posts:
                # 비즈니스 로직
                pass

            offset += chunk_size

트랜잭션 관리

transaction_management.py
from contextlib import contextmanager
from sqlalchemy.orm import Session
from database import SessionLocal

@contextmanager
def database_transaction():
    """트랜잭션 자동 관리"""
    db: Session = SessionLocal()
    try:
        yield db
        db.commit()
    except Exception:
        db.rollback()
        raise
    finally:
        db.close()

def transfer_posts_between_users(from_user_id: int, to_user_id: int):
    """사용자 간 포스트 이전 (트랜잭션 예시)"""

    with database_transaction() as db:
        # 여러 작업을 하나의 트랜잭션으로 처리
        posts = db.query(Post).filter(Post.author_id == from_user_id).all()

        for post in posts:
            post.author_id = to_user_id

        # 로그 기록
        from models import TransferLog
        log = TransferLog(
            from_user_id=from_user_id,
            to_user_id=to_user_id,
            post_count=len(posts)
        )
        db.add(log)

        # 모든 작업이 성공하면 자동 커밋, 실패하면 자동 롤백

이 가이드를 통해 FastAPI와 SQLAlchemy를 활용한 강력하고 확장 가능한 데이터베이스 기반 웹 애플리케이션을 구축할 수 있습니다. 개발 초기에는 간단한 구조로 시작하고, 애플리케이션이 성장함에 따라 고급 기능들을 점진적으로 도입하는 것을 권장합니다.

Comments