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

안녕하세요! 이번 포스팅에서는 FastAPI 애플리케이션에 데이터베이스를 연동하는 방법을 자세히 다루어보겠습니다. 구체적으로는 SQLAlchemy를 사용해 ORM(Object Relational Mapping)을 구성하고, Alembic을 활용하여 데이터베이스 스키마 변경(마이그레이션)을 체계적으로 관리하는 과정을 살펴보겠습니다.

데이터베이스 연동은 백엔드 애플리케이션의 핵심 기능 중 하나입니다. 단순히 데이터를 저장하고 읽어오는 것을 넘어, 스키마 관리와 마이그레이션 전략이 제대로 갖추어져 있어야 프로젝트가 확장될 때도 유지보수가 편리합니다. 이번 글을 통해 FastAPI에서 DB 연동을 어떻게 구성하는지 실제 활용 사례를 예시로 들어 설명해드리겠습니다.

---

3.1. SQLAlchemy와 Alembic 개요

3.1.1. SQLAlchemy란?

• SQLAlchemy는 Python에서 가장 널리 쓰이는 ORM 라이브러리로, 데이터베이스를 객체 지향적으로 다룰 수 있게 해줍니다.
• 개발자가 직접 SQL 쿼리를 작성하지 않고, Python 클래스와 메서드로 DB 연산을 추상화하여 코드를 유지보수하기 쉽습니다.
• 대표적으로 Declarative Base 방식을 사용해 테이블(모델)을 정의하고, 세션(Session)을 통해 CRUD(Create, Read, Update, Delete) 작업을 수행할 수 있습니다.

3.1.2. Alembic이 필요한 이유

• 프로젝트가 진행되면서 DB 스키마(테이블 구조)는 계속 진화합니다. 새로운 컬럼을 추가하거나, 기존 컬럼을 변경/삭제할 수 있습니다.
• Alembic은 SQLAlchemy 팀에서 개발한 DB 마이그레이션 도구로, 스키마 변경 이력을 안전하게 관리하고 여러 단계로 되돌릴 수 있습니다.
• 많은 팀이 “DDL 구문(ALTER TABLE …)”을 수동으로 작성하던 과거와 달리, Alembic은 자동/반자동으로 마이그레이션 파일을 생성해 주므로 생산성이 크게 향상됩니다.

---

3.2. 프로젝트 예시 시나리오

아래 예시는 간단한 전자상거래(E-Commerce) 애플리케이션을 가정하겠습니다.

• User 모델: 사용자 정보(이메일, 비밀번호, 권한 등)를 관리
• Product 모델: 상품 정보(이름, 가격, 재고)를 관리
• Order 모델: 주문 내역을 관리하고, 주문과 유저/상품 간 관계를 설정

물론 실제 상거래 시스템은 훨씬 복잡하겠지만, 본문에서는 기본적인 테이블 연동과 마이그레이션 개념을 익히는 데 초점을 맞추겠습니다.

---

3.3. 의존성 설치 및 설정

3.3.1. Poetry로 패키지 설치

poetry add sqlalchemy alembic psycopg2

• sqlalchemy: ORM 라이브러리
• alembic: DB 마이그레이션 도구
• psycopg2: PostgreSQL 연결 드라이버 (PostgreSQL을 사용한다고 가정)

TIP: MySQL을 사용한다면 pymysql이나 mysqlclient를, SQLite를 사용한다면 별도의 드라이버 없이 sqlite:// URL만으로 가능합니다.

3.3.2. .env 파일 설정

다음과 같이 환경 변수로 DB 연결 정보를 관리합니다.

# .env (예시)
DB_HOST=localhost
DB_PORT=5432
DB_USER=myuser
DB_PASSWORD=mypass
DB_NAME=mydb

이후 load_dotenv() 등을 활용해 Python 코드에서 가져와 사용합니다.

---

3.4. 데이터베이스 세션(Session) 구성

FastAPI 애플리케이션에서 DB 연동의 핵심은 세션(Session) 관리입니다. 일반적으로는 “세션 스코프”를 어떻게 설정하느냐가 중요합니다.

3.4.1. DB 세션 초기화 (db/session.py 예시)

# app/db/session.py
import os
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker

DATABASE_URL = (
    f"postgresql://{os.getenv('DB_USER')}:{os.getenv('DB_PASSWORD')}"
    f"@{os.getenv('DB_HOST')}:{os.getenv('DB_PORT')}/{os.getenv('DB_NAME')}"
)

engine = create_engine(DATABASE_URL, pool_pre_ping=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

• create_engine()로 DB 연결 엔진을 생성
• sessionmaker()를 통해 각 요청마다 세션을 열고 닫을 수 있도록 합니다.
• autocommit=False, autoflush=False는 일반적인 패턴이지만, 필요에 따라 조정 가능합니다.

3.4.2. Base 모델 정의 (db/base.py)

# app/db/base.py
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

• Declarative Base를 생성해, 이후에 정의할 모델들이 이를 상속하여 테이블 정보를 갖도록 합니다.

3.4.3. 의존성 주입(Dependency)로 세션 관리 (api/deps.py 예시)

# app/api/deps.py
from app.db.session import SessionLocal
from typing import Generator

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

• FastAPI의 의존성 주입(Dependency Injection)을 활용하여 라우터 함수마다 DB 세션을 공유합니다.
• 요청이 끝나면 finally에서 db.close()가 호출되어 리소스를 반납합니다.

---

3.5. 모델(Model) 정의와 관계 설정

이제 User, Product, Order 모델을 예시로 살펴봅시다.

3.5.1. User 모델 (models/user.py)

# app/models/user.py
from sqlalchemy import Column, Integer, String
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", "user"

• __tablename__은 실제 DB 테이블 이름입니다.
• email, hashed_password 등 필드를 정의합니다.
• nullable=False, unique=True, index=True 등 제약조건을 부여합니다.

3.5.2. Product 모델 (models/product.py)

# app/models/product.py
from sqlalchemy import Column, Integer, String, Numeric
from app.db.base import Base

class Product(Base):
    __tablename__ = "products"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String, nullable=False)
    description = Column(String, nullable=True)
    price = Column(Numeric(10, 2), nullable=False)
    stock = Column(Integer, default=0)

• 가격은 Numeric(또는 DECIMAL) 타입을 권장합니다.
• 재고(stock)를 표현하고, 기본값을 0으로 설정합니다.

3.5.3. Order 모델 (models/order.py)

# app/models/order.py
from sqlalchemy import Column, Integer, ForeignKey, DateTime, func
from sqlalchemy.orm import relationship
from app.db.base import Base

class Order(Base):
    __tablename__ = "orders"

    id = Column(Integer, primary_key=True, index=True)
    user_id = Column(Integer, ForeignKey("users.id"), nullable=False)
    created_at = Column(DateTime(timezone=True), server_default=func.now())

    user = relationship("User", backref="orders")

• 사용자와의 관계를 설정하기 위해 ForeignKey("users.id")를 지정합니다.
• relationship("User")를 통해 SQLAlchemy ORM 관계를 정의합니다.
• 실제로 OrderItem(주문 상세) 모델도 있을 수 있지만, 예시에서는 간단히 Order만 다루겠습니다.

---

3.6. Alembic으로 마이그레이션 관리

3.6.1. Alembic 초기화

alembic init alembic

• 프로젝트 루트 디렉토리에 alembic/ 폴더가 생성되고, alembic.ini 설정 파일이 생깁니다.

3.6.2. alembic.ini 및 env.py 설정

• alembic.ini에서 DB URL을 직접 넣는 대신, env.py에서 DATABASE_URL을 가져와 쓰도록 구성할 수 있습니다.
• env.py 내에서 from app.db.base import Base 로드 후 target_metadata = Base.metadata 지정하면, Alembic이 모델 기반으로 마이그레이션을 생성합니다.

# alembic/env.py (중요 부분 발췌)
from logging.config import fileConfig
from sqlalchemy import engine_from_config, pool
from app.db.base import Base
import os

config = context.config
fileConfig(config.config_file_name)

target_metadata = Base.metadata

def run_migrations_offline():
    url = os.getenv("DATABASE_URL")
    context.configure(
        url=url, target_metadata=target_metadata, literal_binds=True
    )
    with context.begin_transaction():
        context.run_migrations()

def run_migrations_online():
    config_section = config.get_section(config.config_ini_section)
    config_section["sqlalchemy.url"] = os.getenv("DATABASE_URL")
    connectable = engine_from_config(
        config_section,
        prefix="sqlalchemy.",
        poolclass=pool.NullPool,
    )
    with connectable.connect() as connection:
        context.configure(connection=connection, target_metadata=target_metadata)
        with context.begin_transaction():
            context.run_migrations()

3.6.3. 자동 마이그레이션 파일 생성

# 새 마이그레이션 리비전 생성
alembic revision --autogenerate -m "create user, product, order tables"

• Alembic은 현재 모델(Base.metadata)과 DB 상태를 비교해, 변경 사항을 자동 반영한 스크립트를 생성합니다.
• alembic/versions/ 폴더에 파일이 생깁니다.

3.6.4. 업그레이드(Up) & 다운그레이드(Down)

# DB에 마이그레이션 적용
alembic upgrade head

# 이전 버전으로 되돌리기
alembic downgrade -1

• head는 가장 최신 리비전을 의미합니다.
• -1은 한 단계 이전으로 되돌린다는 의미입니다.

---

3.7. CRUD 예시: User 테이블 활용

3.7.1. CRUD 함수를 정의하는 이유

ORM을 사용하더라도, 일반적으로 “CRUD 레이어”를 모듈화하여 관리합니다. 이렇게 하면 라우터(endpoints) 코드가 깔끔해지고, 테스트 시에도 함수 단위로 검증이 용이합니다.

3.7.2. CRUD 예시 코드 (crud/user.py)

# app/crud/user.py
from sqlalchemy.orm import Session
from app.models.user import User
from app.schemas.user import UserCreate, UserUpdate

def create_user(db: Session, user_in: UserCreate) -> User:
    db_user = User(
        email=user_in.email,
        hashed_password=user_in.password,  # 실서비스에서는 해싱 필수
        role=user_in.role or "user"
    )
    db.add(db_user)
    db.commit()
    db.refresh(db_user)
    return db_user

def get_user_by_id(db: Session, user_id: int) -> User:
    return db.query(User).filter(User.id == user_id).first()

def update_user(db: Session, db_user: User, user_in: UserUpdate) -> User:
    if user_in.email:
        db_user.email = user_in.email
    if user_in.password:
        db_user.hashed_password = user_in.password  # 해싱 가정
    db.commit()
    db.refresh(db_user)
    return db_user

def delete_user(db: Session, db_user: User) -> None:
    db.delete(db_user)
    db.commit()

• db.refresh(db_user): commit 후 최신 상태로 갱신합니다.
• **UserCreate, UserUpdate**는 Pydantic 모델이 될 것이며, password는 실제로는 해싱 처리를 거쳐야 안전합니다.

3.7.3. 라우터에서 사용하기 (api/v1/endpoints/user.py 예시)

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.api.deps import get_db
from app.crud.user import create_user, get_user_by_id, update_user, delete_user
from app.schemas.user import UserCreate, UserUpdate, UserRead

router = APIRouter()

@router.post("/", response_model=UserRead)
def create_user_endpoint(
    user_in: UserCreate,
    db: Session = Depends(get_db)
):
    existing_user = db.query(User).filter(User.email == user_in.email).first()
    if existing_user:
        raise HTTPException(status_code=400, detail="Email already registered")
    
    return create_user(db, user_in)

@router.get("/{user_id}", response_model=UserRead)
def get_user_endpoint(
    user_id: int,
    db: Session = Depends(get_db)
):
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return db_user

@router.put("/{user_id}", response_model=UserRead)
def update_user_endpoint(
    user_id: int,
    user_in: UserUpdate,
    db: Session = Depends(get_db)
):
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    return update_user(db, db_user, user_in)

@router.delete("/{user_id}")
def delete_user_endpoint(
    user_id: int,
    db: Session = Depends(get_db)
):
    db_user = get_user_by_id(db, user_id)
    if not db_user:
        raise HTTPException(status_code=404, detail="User not found")
    delete_user(db, db_user)
    return {"detail": "User deleted"}

• API 경로: /api/v1/users 라우터라고 가정
• UserRead 스키마로 응답 형식을 정의하면, 자동으로 문서(/docs)에도 반영됩니다.

---

3.8. 실제 활용 사례 & 모범 사례

3.8.1. 전자상거래 서비스

• 사용자 정보, 상품 정보, 주문 정보가 각기 다른 테이블에 저장되고, 여러 개의 CRUD 레이어(예: crud/product.py, crud/order.py)를 작성합니다.
• 주문 시 재고 관리나 결제 승인 같은 복잡한 로직은 “서비스(Service) 레이어”를 두어 관리하기도 합니다.

3.8.2. 블로그나 커뮤니티 웹사이트

• 사용자, 게시글, 댓글, 좋아요(Like) 등 다양한 엔티티가 존재하며, Alembic 마이그레이션을 통해 새로운 기능(예: 태그, 첨부파일, 멘션)을 추가할 때마다 스키마를 업그레이드합니다.
• 스키마가 변경되더라도 DB에서 기존 데이터가 안전하게 보존되고, 운영 환경에서도 다운타임 없이 스키마를 진화시킬 수 있습니다.

3.8.3. 사내 ERP 시스템

• 직원 정보, 부서 정보, 급여 계산 등 기업 내부 프로세스를 관리합니다.
• 권한(Role)별로 접근할 수 있는 테이블과 컬럼이 달라지므로, ORM을 쓰면서도 세부 권한 로직은 추가적으로 구현하게 됩니다.

모범 사례

1. 명확한 테이블 네이밍: 테이블/컬럼 명을 일관되고 직관적으로 작성합니다.
2. 인덱스(index) 설계: 자주 조회되는 칼럼은 인덱스를 걸어 성능을 개선합니다.
3. 트랜잭션 관리: 여러 DB 작업이 하나의 논리적 단위로 처리되어야 할 때는 트랜잭션을 사용하는 것이 안전합니다.
4. 데이터베이스 별 특성 고려: PostgreSQL, MySQL, SQLite는 미묘하게 다른 기능들을 제공하므로, 프로젝트 규모와 요구사항에 맞춰 선택합니다.

---

3.9. 마무리

이번 3장에서는 FastAPI 애플리케이션과 SQLAlchemy, Alembic을 이용해 데이터베이스를 연동하고 CRUD를 구축하는 전반적인 흐름을 배웠습니다. 구체적으로는:

1. SQLAlchemy로 ORM 모델을 정의하고 세션을 구성하는 방법
2. Alembic으로 DB 스키마 변경 이력을 안전하게 관리하는 방법
3. CRUD 함수를 작성하고 라우터에 연동하는 패턴
4. 실제 전자상거래, 블로그 등의 활용 사례

다음 장에서는 구현한 DB 모델을 기반으로 더욱 구체적인 API 설계와 Pydantic 스키마 활용, 나아가 인증/권한 처리까지 이어집니다. 데이터 저장 및 관리라는 토대를 확실히 다졌으니, 이제 본격적으로 기능을 확장해 봅시다!