들어가며
소규모 웹 애플리케이션을 개발하기 시작할 때는 모든 것이 단순합니다. 하나의 데이터베이스로 개발도 하고, 테스트도 하고, 심지어 프로덕션까지 운영하는 경우가 많습니다. 처음에는 이 접근법이 효율적으로 보입니다. "왜 복잡하게 만들어야 하지?" 라는 질문이 합리적이게 느껴집니다. 그러나 애플리케이션이 조금만 커져도 단순함이 점차 걸림돌로 변하는 것을 느껴보셨을 것입니다. 개발자들은 몇 가지 현실적인 문제에 마주하게 됩니다:
•
"새로운 기능을 개발하려면 테이블 구조를 변경해야 하는데, 실제 사용자 데이터가 있는 프로덕션 DB를 어떻게 안전하게 수정할 수 있을까?"
•
"DB와 상호작용하는 로직의 테스트/프로덕션 소스코드의 중복, DB 마이그레이션 과정에서의 불필요한 수작업을 줄일 방법은 없을까?“
•
"개발 환경, 테스트 환경, 프로덕션 환경에서 서로 다른 데이터베이스를 사용해야 한다는 것은 알겠는데, 도대체 코드를 어떻게 구성해야 할까?"
이러한 질문들은 단순히 기술적인 도전을 넘어 프로젝트의 지속 가능성과 팀의 생산성에 직접적인 영향을 미칩니다. 테스트를 올바르게 적용하기 어려운 소스코드의 확장은 점점 느려집니다. 잘못된 데이터베이스 변경 하나가 서비스 중단이나 데이터 손실로 이어질 수 있으며, 실제 사용자가 있는 서비스라면 신뢰 하락과 금전적 손실로 이어질 수 있겠지요.
특히 FastAPI와 같은 현대적인 프레임워크로 빠르게 개발을 진행하다 보면, 데이터베이스 관리 측면에서의 체계적인 접근법이 부족한 채로 기술 부채가 쌓여가는 경우가 많습니다. 이 글에서는 FastAPI 애플리케이션에서 데이터베이스 스키마를 안전하게 관리하고, 테스트와 프로덕션 환경을 효과적으로 분리하는 방법을 살펴보려 합니다. 구체적으로:
1.
데이터베이스 마이그레이션: Alembic을 활용하여 스키마 변경을 체계적으로 관리하고 버전 관리하는 방법
2.
테스트-프로덕션 환경 분리: 테스트 환경에서 안전하게 변경사항을 검증한 후 프로덕션에 적용하는 워크플로우
3.
의존성 주입을 통한 환경 전환: FastAPI의 강력한 의존성 주입 시스템을 활용하여 코드 변경 없이 다양한 환경 간에 전환하는 방법
이러한 접근법을 통해 데이터베이스 관련 작업이 더 이상 위험하고 스트레스가 많은 일이 아닌, 예측 가능하고 신뢰할 수 있는 프로세스로 변화하는 과정을 살펴보겠습니다. 애플리케이션이 성장함에 따라 발생하는 '성장통'을 관리하고, 더 견고하고 유지보수 가능한 시스템을 구축하는 데 필요한 실질적인 지식을 공유하고자 합니다. 언어모델을 이용해 구체화하는 시간이 매우 빨라진 오늘날, 테스트 가능한 DB를 세팅하는 워크플로와 코드 스니펫을 한 번 숙지해 두면 잠재적으로 리소스를 많이 절약할 수 있으리라 믿습니다.
DB 마이그레이션 워크플로
데이터베이스 스키마는 데이터가 어떻게 구성되고 저장되는지를 정의하는 구조입니다. 쉽게 말해 테이블, 필드, 관계, 제약 조건 등을 포함한 데이터베이스의 청사진이라고 볼 수 있습니다. SQLAlchemy를 사용하면 Python 클래스로 이러한 스키마를 정의하게 됩니다.
예를 들어, 간단한 사용자 테이블의 스키마를 SQLAlchemy 모델로 정의하면 다음과 같습니다:
from sqlalchemy import Column, Integer, String, DateTime
from sqlalchemy.ext.declarative import declarative_base
from datetime import datetime
Base = declarative_base()
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
name = Column(String(100), nullable=False)
email = Column(String(100), unique=True, nullable=False)
created_at = Column(DateTime, default=datetime.utcnow)
Python
복사
이 코드는 id, name, email, created_at 필드를 가진 users 테이블을 정의합니다. 애플리케이션을 처음 설정할 때는 이러한 모델 클래스를 정의하고 데이터베이스에 테이블을 생성하는 것으로 충분합니다.
그러나 시간이 지나면서 비즈니스 요구사항이 변경되고 이에 따라 데이터베이스 스키마도 변경해야 할 필요가 생깁니다. 예를 들어, 사용자 이름을 first_name과 last_name으로 분리하거나, 전화번호 필드를 추가하는 등의 변경이 필요할 수 있습니다.
이때 마이그레이션이 필요합니다. 데이터베이스 마이그레이션은 한 상태에서 다른 상태로 데이터베이스 스키마를 변경하는 과정입니다. 이는 단순히 테이블 구조를 변경하는 것뿐만 아니라, 기존 데이터를 새 구조에 맞게 변환하는 작업도 포함합니다. SQLAlchemy와 함께 사용할 수 있는 Alembic이라는 도구는 이러한 마이그레이션 과정을 관리하는 데 도움을 줍니다. 먼저 Alembic을 설치합니다:
pip install alembic
Shell
복사
설치 후, 프로젝트에 Alembic 환경을 초기화합니다:
alembic init migrations
Shell
복사
이 명령어는 'migrations' 디렉토리와 'alembic.ini' 파일을 생성합니다. 다음으로 데이터베이스 연결 정보를 설정해야 합니다. 'alembic.ini' 파일에서 데이터베이스 URL을 설정합니다:
# alembic.ini
sqlalchemy.url = postgresql://username:password@localhost/dbname
Plain Text
복사
그리고 'migrations/env.py' 파일에서 SQLAlchemy 모델을 Alembic과 연결합니다:
# migrations/env.py (일부)
import sys
from pathlib import Path
# 모델이 정의된 패키지를 sys.path에 추가
sys.path.append(str(Path(__file__).parent.parent))
from app.models import Base # 당신의 모델 Base import
target_metadata = Base.metadata
Python
복사
이제 User 모델에 변경이 필요하다고 가정해 봅시다. 예를 들어, name 필드를 first_name과 last_name으로 분리하고 phone_number 필드를 추가하려면, 먼저 모델을 다음과 같이 변경합니다:
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
# name 필드 대신 first_name과 last_name으로 분리
first_name = Column(String(50), nullable=False)
last_name = Column(String(50), nullable=False)
email = Column(String(100), unique=True, nullable=False)
# 새 필드 추가
phone_number = Column(String(20), nullable=True)
created_at = Column(DateTime, default=datetime.utcnow)
Python
복사
모델 변경 후, Alembic을 사용하여 마이그레이션 스크립트를 생성합니다:
alembic revision \
--autogenerate -m "Split name into first_name and last_name, add phone_number"
Shell
복사
이 명령어는 모델과 데이터베이스 상태를 비교하여 변경 사항을 감지하고, 'migrations/versions/' 디렉토리에 마이그레이션 스크립트를 생성합니다. 생성된 스크립트는 다음과 같을 수 있습니다:
# migrations/versions/a1b2c3d4e5f6_split_name_into_first_name_and_last_name.py
"""Split name into first_name and last_name, add phone_number
Revision ID: a1b2c3d4e5f6
Revises: 9z8y7x6w5v4
Create Date: 2023-12-01 12:34:56.789012
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers
revision = 'a1b2c3d4e5f6'
down_revision = '9z8y7x6w5v4'
branch_labels = None
depends_on = None
def upgrade():
# 새 컬럼 추가
op.add_column('users', sa.Column('first_name', sa.String(50), nullable=True))
op.add_column('users', sa.Column('last_name', sa.String(50), nullable=True))
op.add_column('users', sa.Column('phone_number', sa.String(20), nullable=True))
# 여기서는 기존 데이터 마이그레이션 로직이 필요합니다!
# 기존 name 컬럼 삭제
op.drop_column('users', 'name')
def downgrade():
# 롤백 로직 (역순)
op.add_column('users', sa.Column('name', sa.String(100), nullable=True))
# 여기서는 first_name과 last_name을 name으로 되돌리는 로직이 필요합니다!
op.drop_column('users', 'first_name')
op.drop_column('users', 'last_name')
op.drop_column('users', 'phone_number')
Python
복사
그러나 이 자동 생성된 스크립트는 기존 데이터를 처리하는 로직이 없습니다. 이를 수정하여 데이터 마이그레이션 로직을 추가해야 합니다:
def upgrade():
# 새 컬럼 추가
op.add_column('users', sa.Column('first_name', sa.String(50), nullable=True))
op.add_column('users', sa.Column('last_name', sa.String(50), nullable=True))
op.add_column('users', sa.Column('phone_number', sa.String(20), nullable=True))
# 기존 데이터 마이그레이션
# 임시로 사용할 테이블 참조 생성
users = sa.table('users',
sa.column('id', sa.Integer),
sa.column('name', sa.String),
sa.column('first_name', sa.String),
sa.column('last_name', sa.String)
)
# 데이터베이스 연결 가져오기
connection = op.get_bind()
# 기존 사용자 데이터 조회
results = connection.execute(sa.select([users.c.id, users.c.name]))
# 각 사용자의 name을 first_name과 last_name으로 분리
for user_id, name in results:
if name:
# 공백을 기준으로 이름 분리
name_parts = name.split(' ', 1)
first_name = name_parts[0]
last_name = name_parts[1] if len(name_parts) > 1 else ''
# 분리된 이름으로 업데이트
connection.execute(
users.update().
where(users.c.id == user_id).
values(first_name=first_name, last_name=last_name)
)
# 이제 first_name과 last_name이 채워졌으니 NOT NULL 제약 조건 추가
op.alter_column('users', 'first_name', nullable=False)
op.alter_column('users', 'last_name', nullable=False)
# 마지막으로 기존 name 컬럼 삭제
op.drop_column('users', 'name')
Python
복사
이 업그레이드 함수는 다음 단계로 진행됩니다:
1.
먼저 새 컬럼들을 nullable로 추가합니다.
2.
기존 'name' 데이터를 조회하고 이를 'first_name'과 'last_name'으로 분리합니다.
3.
분리된 데이터로 새 컬럼을 업데이트합니다.
4.
데이터 마이그레이션이 완료된 후 필요한 제약 조건(NOT NULL)을 추가합니다.
5.
마지막으로 기존 'name' 컬럼을 삭제합니다.
마이그레이션 스크립트가 준비되면 다음 명령어로 데이터베이스를 업그레이드합니다:
alembic upgrade head
Shell
복사
이 명령어는 가장 최신 버전으로 데이터베이스를 업그레이드합니다. 단계적으로 진행하려면 다음과 같이 실행할 수 있습니다:
alembic upgrade +1 # 한 단계만 업그레이드
Shell
복사
문제가 발생하면 다음 명령어로 롤백할 수 있습니다:
alembic downgrade -1 # 한 단계 롤백
Shell
복사
대규모 데이터베이스의 경우, 대량의 레코드를 마이그레이션하는 것은 시간이 오래 걸리고 메모리를 많이 사용할 수 있습니다. 이런 경우 배치 처리를 사용하여 성능을 개선할 수 있습니다:
# 대용량 데이터 배치 처리 예시
batch_size = 1000
offset = 0
while True:
batch = connection.execute(
sa.select([users.c.id, users.c.name])
.limit(batch_size)
.offset(offset)
).fetchall()
if not batch:
break # 더 이상 처리할 데이터가 없으면 종료
for user_id, name in batch:
# 이름 분리 및 업데이트 로직
offset += batch_size
print(f"Processed {offset} records")
Python
복사
마이그레이션을 자동화하려면 배포 스크립트나 CI/CD 파이프라인에 다음과 같은 단계를 추가할 수 있습니다:
#!/bin/bash
# 배포 스크립트 예시
# 코드 업데이트
git pull
# 의존성 설치
pip install -r requirements.txt
# 데이터베이스 마이그레이션
alembic upgrade head
# 애플리케이션 재시작
uvicorn app.main:app --reload
Shell
복사
이렇게 Alembic을 사용하면 데이터베이스 스키마 변경을 체계적으로 관리하고, 기존 데이터를 안전하게 마이그레이션할 수 있습니다. 각 변경 사항은 버전 관리되며, 필요할 경우 이전 상태로 롤백할 수도 있습니다.
시간이 지나면서 마이그레이션 히스토리가 쌓이게 되면, 이는 데이터베이스 스키마의 변화 과정을 문서화하는 역할도 합니다. 새로운 팀원이 합류하거나, 오래된 변경 사항을 추적해야 할 때 이 히스토리는 매우 유용한 자료가 됩니다. Alembic의 마이그레이션 스크립트는 코드로 작성되어 버전 관리 시스템에 저장되므로, 코드 변경과 데이터베이스 변경의 이력을 함께 관리할 수 있다는 장점도 있습니다.
Test DB가 포함된 워크플로
프로젝트가 성장하면서 데이터베이스 관리 전략도 함께 발전해야 합니다. 처음에는 단일 데이터베이스로 개발과 테스트, 프로덕션까지 모두 해결하던 방식이 점차 한계에 부딪히게 됩니다. 특히 프로덕션 데이터가 쌓이고 서비스가 운영되면서 직접 프로덕션 데이터베이스를 수정하는 것은 상당한 부담과 위험을 수반하게 됩니다. 이런 상황에서는 테스트용 데이터베이스와 프로덕션 데이터베이스를 분리하여 안전하게 변경 사항을 적용하는 전략이 필요합니다.
먼저 환경별로 데이터베이스 연결 설정을 관리할 수 있는 구조를 마련해야 합니다. FastAPI 프로젝트에서는 Pydantic과 환경 변수를 활용하여 이를 구현할 수 있습니다. 예를 들어, 프로젝트 루트에 config.py 파일을 만들고 다음과 같이 설정 클래스를 정의할 수 있습니다:
# app/config.py
from pydantic import BaseSettings
class Settings(BaseSettings):
DATABASE_URL: str
ENVIRONMENT: str = "dev" # 기본값은 개발 환경
class Config:
env_file = ".env"
# 환경 변수나 .env 파일에서 설정 로드
settings = Settings()
Python
복사
그리고 각 환경별로 다른 .env 파일을 만들어 관리합니다. 테스트 환경에서는 .env.test를, 프로덕션 환경에서는 .env.prod를 사용하는 식입니다. 이렇게 하면 코드 변경 없이 환경 변수만으로 연결할 데이터베이스를 쉽게 전환할 수 있습니다.
이제 개발자 수민이 새로운 기능을 개발하는 상황을 생각해 봅시다. 수민은 사용자 모델에 주소 정보를 추가해야 합니다. 기존에는 사용자 모델에 이름과 이메일만 있었지만, 이제 배송 기능을 추가하면서 주소 정보가 필요해졌습니다.
수민은 먼저 로컬 개발 환경에서 모델을 수정합니다:
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True)
first_name = Column(String(50), nullable=False)
last_name = Column(String(50), nullable=False)
email = Column(String(100), unique=True, nullable=False)
# 새로 추가된 필드들
address_street = Column(String(100), nullable=True)
address_city = Column(String(50), nullable=True)
address_country = Column(String(50), nullable=True)
Python
복사
모델을 변경한 후, 수민은 테스트 환경에서 마이그레이션을 생성하고 테스트하기로 합니다. 먼저 테스트 환경 설정을 활성화합니다:
export ENV_FILE=.env.test
Shell
복사
이렇게 하면 앞서 설정한 config.py가 .env.test 파일에서 설정을 로드하게 되고, 테스트 데이터베이스에 연결됩니다. 이제 수민은 Alembic을 사용하여 마이그레이션 스크립트를 생성합니다:
alembic revision --autogenerate -m "Add address fields to users"
Shell
복사
이 명령어는 현재 모델과 테스트 데이터베이스의 스키마를 비교하여 변경 사항을 감지하고, 자동으로 마이그레이션 스크립트를 생성합니다. 생성된 스크립트는 migrations/versions/ 디렉토리에 저장됩니다. 수민은 이 스크립트를 열어 내용을 검토합니다:
"""Add address fields to users
Revision ID: abc123def456
Revises: 98765zyx432
Create Date: 2023-12-15 10:30:45.123456
"""
from alembic import op
import sqlalchemy as sa
# revision identifiers
revision = 'abc123def456'
down_revision = '98765zyx432'
branch_labels = None
depends_on = None
def upgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.add_column('users', sa.Column('address_street', sa.String(100), nullable=True))
op.add_column('users', sa.Column('address_city', sa.String(50), nullable=True))
op.add_column('users', sa.Column('address_country', sa.String(50), nullable=True))
# ### end Alembic commands ###
def downgrade():
# ### commands auto generated by Alembic - please adjust! ###
op.drop_column('users', 'address_country')
op.drop_column('users', 'address_city')
op.drop_column('users', 'address_street')
# ### end Alembic commands ###
Python
복사
이 경우에는 단순히 컬럼을 추가하는 것이므로 자동 생성된 스크립트가 적절해 보입니다. 만약 더 복잡한 데이터 변환이 필요하다면 스크립트를 수정할 수도 있습니다. 수민은 이제 테스트 데이터베이스에 마이그레이션을 적용합니다:
alembic upgrade head
Shell
복사
마이그레이션이 성공적으로 적용되면, 수민은 수정된 모델을 사용하는 코드를 작성하고 테스트합니다. 새로운 주소 필드를 사용하는 API 엔드포인트를 구현하고, 단위 테스트와 통합 테스트를 작성하여 기능이 제대로 작동하는지 확인합니다:
pytest tests/test_user_address.py
Shell
복사
테스트가 모두 통과하면, 수민은 변경 사항을 코드 저장소에 커밋하고 팀의 코드 리뷰를 요청합니다. 이때 마이그레이션 스크립트도 함께 커밋하여 다른 팀원들이 동일한 스키마 변경을 적용할 수 있게 합니다.
코드 리뷰가 완료되고 모든 CI 테스트가 통과하면, 이제 프로덕션 환경에 변경 사항을 적용할 차례입니다. 그러나 프로덕션 데이터베이스에는 이미 많은 사용자 데이터가 있으므로, 변경 사항을 적용하기 전에 추가적인 검증이 필요합니다.
이를 위해 팀은 프로덕션 데이터의 일부를 테스트 환경으로 가져와 실제 마이그레이션 스크립트를 테스트합니다. 이렇게 하면 프로덕션 데이터의 특성과 다양성을 고려한 실제적인 검증이 가능해집니다:
#!/bin/bash
# scripts/test_prod_data_migration.sh
# 1. 프로덕션 DB에서 데이터 덤프 생성 (민감 정보는 제외하거나 마스킹)
echo "Creating anonymized dump from production database..."
pg_dump -h $PROD_DB_HOST -U $PROD_DB_USER -d $PROD_DB_NAME \\
--table=users --data-only \\
--column-inserts | sed 's/\\(email=\\).*,/\\1'\\''anonymized@example.com'\\'',/g' > anonymized_users.sql
# 2. 테스트 DB 재설정
echo "Resetting test database to pre-migration state..."
export ENV_FILE=.env.test
alembic downgrade base # 모든 마이그레이션 되돌리기
# 3. 테스트 DB의 스키마를 마이그레이션 직전 상태로 설정
echo "Setting up test database schema..."
alembic upgrade abc123def456~1 # 새 마이그레이션 바로 이전 버전으로 설정
# 4. 익명화된 프로덕션 데이터 로드
echo "Loading anonymized production data..."
psql -h $TEST_DB_HOST -U $TEST_DB_USER -d $TEST_DB_NAME < anonymized_users.sql
# 5. 실제 마이그레이션 스크립트 적용
echo "Applying migration to production-like data..."
alembic upgrade +1 # 한 단계 업그레이드 (새 마이그레이션 적용)
# 6. 마이그레이션 결과 검증
echo "Validating migration results..."
psql -h $TEST_DB_HOST -U $TEST_DB_USER -d $TEST_DB_NAME \\
-c "SELECT COUNT(*) FROM users WHERE address_street IS NULL;" \\
-c "SELECT COUNT(*) FROM users WHERE address_street IS NOT NULL;"
# 7. 롤백 테스트
echo "Testing rollback..."
alembic downgrade -1 # 마이그레이션 롤백
# 8. 롤백 결과 검증
echo "Validating rollback results..."
psql -h $TEST_DB_HOST -U $TEST_DB_USER -d $TEST_DB_NAME \\
-c "SELECT COUNT(*) FROM information_schema.columns WHERE table_name='users' AND column_name='address_street';"
echo "Migration test with production data completed."
Shell
복사
그런데 잠깐, 프로덕션 데이터를 사용해 마이그레이션을 테스트해보는 이유는 무엇일까요? 가장 중요한 것은 엣지 케이스를 발견할 수 있다는 점입니다. 실제 프로덕션 환경에서는 개발자가 예상하지 못한 다양한 데이터 패턴이 존재합니다. 예를 들어, 특수 문자가 포함된 이름, 긴 주소, 국제 문자셋이 포함된 데이터 등이 있을 수 있습니다. 이러한 엣지 케이스는 테스트 환경에서 인위적으로 만든 데이터에서는 발견하기 어렵지만, 실제 마이그레이션에서 문제를 일으킬 수 있습니다.
또한, 실제 데이터를 사용한 테스트는 리소스 사용량을 현실적으로 예측할 수 있게 해줍니다. 몇 백 개의 테스트 레코드에서는 즉시 완료되는 마이그레이션이 수백만 개의 레코드가 있는 프로덕션 환경에서는 몇 시간이 걸릴 수도 있습니다. 이런 성능 특성을 미리 알면 배포 계획을 더 효과적으로 세울 수 있습니다. 예를 들어, 트래픽이 적은 시간대에 배포하거나, 점진적 배포 전략을 채택하는 등의 조치를 취할 수 있습니다.
그런 의미에서 롤백 테스트도 프로덕션 데이터로 진행하는 것이 중요합니다. 마이그레이션 중 문제가 발생했을 때 원래 상태로 돌아갈 수 있는지 확인하는 것은 안전망을 마련하는 것과 같습니다. 특히 데이터 변환이 포함된 복잡한 마이그레이션에서는 롤백 후 데이터가 원래 상태로 정확히 빠르게 복원되는지 검증해야 합니다. 이런 검증 없이 프로덕션 환경에 적용했다가 문제가 발생하면, 롤백 자체가 추가적인 데이터 손상이나 서비스 지연을 일으킬 수 있습니다.
모든 테스트와 검증이 완료되면, 팀은 프로덕션 배포를 위한 구체적인 계획을 수립합니다. 이 계획에는 배포 시점, 필요한 서비스 중단 시간, 담당자 역할, 모니터링 방법, 그리고 문제 발생 시 대응 전략 등이 포함됩니다. 대규모 변경의 경우 사용자에게 미리 공지하여 예상되는 서비스 영향을 알리는 것도 중요합니다.
배포 당일, 운영 담당자 용재은 계획에 따라 다음과 같은 순서로 마이그레이션을 진행합니다. 먼저 프로덕션 환경 설정을 활성화하고, 데이터베이스의 현재 상태를 백업합니다. 그런 다음 현재 마이그레이션 버전을 확인하고, 새 마이그레이션을 적용합니다. 마지막으로 마이그레이션이 성공적으로 적용되었는지 확인하기 위해 새 버전을 다시 확인합니다. 모든 검증이 완료되면, 이제 프로덕션 환경에 마이그레이션을 적용할 차례입니다.
1.
먼저 배포 시간을 사용자들에게 미리 공지합니다 (필요한 경우).
2.
데이터베이스 백업을 생성합니다.
3.
정해진 시간에 마이그레이션을 적용합니다.
4.
애플리케이션을 모니터링하여 문제가 없는지 확인합니다.
# 1. 프로덕션 환경 설정 활성화
export ENV_FILE=.env.prod
# 2. 데이터베이스 백업 생성
pg_dump -h $DB_HOST -U $DB_USER -d $DB_NAME > backup_$(date +%Y%m%d_%H%M%S).sql
# 3. 현재 마이그레이션 버전 확인
alembic current
# 4. 마이그레이션 적용
alembic upgrade head
# 5. 새 버전 확인
alembic current
Shell
복사
마이그레이션이 성공적으로 적용되면, 용재은 애플리케이션 로그와 모니터링 도구를 확인하여 문제가 없는지 확인합니다. 모든 것이 정상이면 배포가 완료된 것입니다.
만약 문제가 발생한다면, 용재은 롤백 계획을 실행할 수 있습니다:
# 마이그레이션 롤백
alembic downgrade -1
# 문제가 지속되면 백업에서 복원
psql -h $DB_HOST -U $DB_USER -d $DB_NAME < backup_20231215_103045.sql
Shell
복사
이러한 과정을 통해 팀은 테스트 환경에서 충분히 검증한 후에 프로덕션 환경에 안전하게 변경 사항을 적용할 수 있습니다. 시간이 지나면서 이 프로세스는 더욱 자동화되고 CI/CD 파이프라인에 통합될 수 있습니다.
이런 방식으로 테스트 데이터베이스와 프로덕션 데이터베이스를 분리하여 관리하면, 개발 과정에서의 유연성은 유지하면서도 프로덕션 환경의 안정성을 보장할 수 있습니다. 새로운 기능을 개발하거나 데이터 모델을 변경할 때, 개발자들은 테스트 환경에서 자유롭게 실험하고 검증할 수 있으며, 검증된 변경 사항만 프로덕션 환경에 적용함으로써 서비스 중단이나 데이터 손실의 위험을 최소화할 수 있습니다.
글에 언급된 워크플로를 정리해 보면 다음과 같습니다:
flowchart TB
%% 프로덕션 데이터 테스트 세부 과정
subgraph prodTest ["프로덕션 데이터 테스트"]
J1[프로덕션 데이터 샘플링] --> J2[데이터 익명화]
J2 --> J3[테스트 DB 준비]
J3 --> J4[마이그레이션 적용]
J4 --> J5[결과 검증]
J5 --> J6[롤백 테스트]
end
style prodTest fill:#e8f8e8,stroke:#50c050,stroke-width:5px
classDef whiteBox fill:white,stroke:#333,stroke-width:1px
class J1,J2,J3,J4,J5,J6 whiteBoxMermaid
복사
flowchart TB
subgraph 개발환경
A[모델 변경] --> B[마이그레이션 스크립트 생성]
B --> C[테스트 DB에 적용]
C --> D[기능 개발 및 테스트]
D --> E{테스트 통과?}
E -->|No| A
E -->|Yes| F[코드 리포지토리에 커밋]
end
subgraph "CI/CD 시스템"
F --> G[자동 테스트 실행]
G --> H{테스트 통과?}
H -->|No| I[개발자에게 알림]
I --> A
H -->|Yes| J[프로덕션 데이터 테스트]
J --> K{검증 통과?}
K -->|No| I
K -->|Yes| L[스테이징 환경에 배포]
end
subgraph "프로덕션 배포"
L --> M[QA 검증]
M --> N{검증 통과?}
N -->|No| O[이슈 보고]
O --> A
N -->|Yes| P[배포 계획 수립]
P --> Q[프로덕션 DB 백업]
Q --> R[마이그레이션 적용]
R --> T{문제 발생?}
T -->|Yes| U[롤백 계획 실행]
U --> V[원인 분석]
V --> A
T -->|No| W[배포 완료]
end
%% 스타일 설정
classDef primary fill:#d0e0ff,stroke:#3080ff,stroke-width:2px
classDef secondary fill:#e8f8e8,stroke:#50c050,stroke-width:2px
classDef special fill:#e8f8e8,stroke:#50c050,stroke-width:5px
classDef warning fill:#fff0d0,stroke:#e0a020,stroke-width:2px
classDef danger fill:#ffe0e0,stroke:#e05050,stroke-width:2px
class A,B,C,D,F primary
class J,L,P,Q,R,S,W secondary
class E,H,K,N,T warning
class I,O,U,V danger
class J1,J2,J3,J4,J5,J6,J specialMermaid
복사
의존성 주입을 이용한 FastAPI 데이터베이스 환경 분리
FastAPI에서 테스트 데이터베이스와 프로덕션 데이터베이스를 효과적으로 분리하기 위해서는 의존성 주입(Dependency Injection) 패턴이 핵심적인 역할을 합니다. FastAPI의 Depends() 기능은 이러한 패턴을 간결하고 강력하게 구현할 수 있게 해주며, 환경에 따라 데이터베이스 연결을 유연하게 전환할 수 있는 기반을 제공합니다.
데이터베이스 연결을 관리하는 가장 기본적인 방법은 다음과 같이 의존성 함수를 정의하는 것입니다:
# app/db/session.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.config import settings
engine = create_engine(settings.DATABASE_URL)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()
def get_db():
db = SessionLocal()
try:
yield db # 여기서 yield가 중요합니다
finally:
db.close()
Python
복사
여기서 주목할 부분은 yield 키워드입니다. 이 간단한 키워드가 FastAPI의 의존성 주입 시스템과 함께 작동하여 HTTP 요청 수명 주기 동안 데이터베이스 연결을 관리하는 중요한 역할을 합니다. 요청이 들어오면 FastAPI는 먼저 의존성 함수를 실행하여 yield 문까지 코드를 실행합니다. 이때 데이터베이스 세션이 생성되고, 이 세션이 엔드포인트 함수에 전달됩니다. 엔드포인트 함수의 실행이 완료된 후(응답이 반환된 후), FastAPI는 다시 의존성 함수로 돌아와 yield 이후의 코드를 실행합니다. 이 경우에는 finally 블록 안에서 데이터베이스 세션을 닫습니다. 이러한 방식으로 FastAPI는 요청이 완료된 후 자동으로 리소스를 정리할 수 있습니다.
그렇다면 왜 단순히 데이터베이스 세션을 함수 인자로 전달하는 대신 Depends()라는 특별한 클래스를 사용할까요? 이것이 바로 FastAPI의 의존성 주입 시스템의 핵심입니다. Depends()는 단순한 인자 전달 이상의 기능을 제공합니다. 가장 중요한 점은 의존성 오버라이딩(dependency overriding)을 가능하게 한다는 것입니다. 이는 코드를 변경하지 않고도 런타임에 의존성의 구현을 교체할 수 있다는 의미입니다. 특히 테스트 환경에서 이 기능은 매우 강력합니다.
API 엔드포인트에서 이러한 의존성을 사용하는 방법은 다음과 같습니다:
@router.post("/users/", response_model=UserResponse)
def create_user(user: UserCreate, db: Session = Depends(get_db)):
db_user = User(
first_name=user.first_name,
last_name=user.last_name,
email=user.email
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
Python
복사
이제 테스트 시에 의존성 오버라이딩이 어떻게 동작하는지 살펴보겠습니다. FastAPI는 app.dependency_overrides 딕셔너리를 통해 의존성 함수를 다른 함수로 교체할 수 있는 메커니즘을 제공합니다:
# tests/conftest.py
@pytest.fixture(scope="function")
def client(db):
# 테스트 중에 get_db 의존성을 오버라이드
def override_get_db():
try:
yield db # 테스트용 데이터베이스 세션 제공
finally:
pass # db.close()는 외부에서 처리됨
# 핵심: 의존성 교체
app.dependency_overrides[get_db] = override_get_db
with TestClient(app) as c:
yield c
# 테스트 후 오버라이드 제거
app.dependency_overrides = {}
Python
복사
이 코드의 핵심은 app.dependency_overrides[get_db] = override_get_db 라인입니다. 이 한 줄로 인해 테스트 중에는 모든 Depends(get_db)가 프로덕션 데이터베이스 대신 테스트 데이터베이스에 연결하게 됩니다. 애플리케이션 코드는 전혀 변경하지 않았지만, 의존성의 구현을 완전히 다른 것으로 교체했습니다. 이것이 바로 Depends()의 강력한 기능입니다(ref1).
더 발전된 아키텍처에서는 리포지토리(Repository) 패턴을 도입하여 데이터 액세스 로직을 캡슐화할 수 있습니다. 리포지토리 패턴은 데이터 소스 접근 로직을 비즈니스 로직으로부터 분리하는 디자인 패턴으로, 코드의 유지보수성과 테스트 용이성을 향상시킵니다. 이는 클린 아키텍처(Clean Architecture)나 육각형 아키텍처(Hexagonal Architecture)와 같은 더 큰 아키텍처 패턴의 일부로 사용되기도 합니다.
리포지토리 계층을 도입하면 데이터 접근 로직을 중앙화하고 추상화할 수 있습니다. 이를 통해 코드 중복이 크게 줄어들고, 비즈니스 로직과 데이터 접근 로직이 명확하게 분리됩니다. 예를 들어, 사용자 정보를 조회하는 로직이 여러 엔드포인트에 반복적으로 나타날 수 있습니다. 이메일로 사용자를 찾는 기능이 로그인, 회원가입, 비밀번호 재설정 등 여러 기능에서 필요하다고 생각해 봅시다.
# 리포지토리 없이 직접 데이터베이스 접근하는 경우:
@router.post("/login/")
def login(credentials: LoginCredentials, db: Session = Depends(get_db)):
user = db.query(User).filter(User.email == credentials.email).first()
if not user or not verify_password(credentials.password, user.hashed_password):
raise HTTPException(status_code=401, detail="Invalid credentials")
# 로그인 로직...
@router.post("/signup/")
def signup(user_data: UserCreate, db: Session = Depends(get_db)):
# 이메일 중복 검사
existing_user = db.query(User).filter(User.email == user_data.email).first()
if existing_user:
raise HTTPException(status_code=400, detail="Email already registered")
# 회원가입 로직...
@router.post("/password-reset/")
def password_reset(request: PasswordResetRequest, db: Session = Depends(get_db)):
user = db.query(User).filter(User.email == request.email).first()
if not user:
# 보안을 위해 사용자가 없어도 성공 응답
return {"message": "Password reset email sent if account exists"}
# 비밀번호 재설정 로직...
Python
복사
위 코드에서 db.query(User).filter(User.email == email).first() 로직이 세 곳에서 반복됩니다. 만약 이메일 조회 방식을 변경해야 한다면(예: 이메일 대소문자 구분 없이 처리), 세 곳 모두 수정해야 합니다. 또한 각 엔드포인트 함수가 데이터베이스 쿼리 방법에 대한 지식을 가지고 있어야 합니다.
리포지토리 패턴을 적용하면 이런 중복을 제거할 수 있습니다:
# app/db/repositories/users.py
class UserRepository(BaseRepository):
def get_by_email(self, email: str):
return self.db.query(User).filter(User.email == email).first()
def create(self, user_data: dict):
user = User(**user_data)
self.db.add(user)
self.db.commit()
self.db.refresh(user)
return user
# app/api/endpoints/users.py
@router.post("/login/")
def login(credentials: LoginCredentials, user_repo: UserRepository = Depends()):
user = user_repo.get_by_email(credentials.email)
if not user or not verify_password(credentials.password, user.hashed_password):
raise HTTPException(status_code=401, detail="Invalid credentials")
# 로그인 로직...
@router.post("/signup/")
def signup(user_data: UserCreate, user_repo: UserRepository = Depends()):
# 이메일 중복 검사
existing_user = user_repo.get_by_email(user_data.email)
if existing_user:
raise HTTPException(status_code=400, detail="Email already registered")
# 회원가입 로직...
Python
복사
이제 이메일로 사용자를 찾는 로직이 UserRepository 클래스의 get_by_email 메서드에 중앙화되었습니다. 나중에 이 로직을 변경해야 한다면 한 곳만 수정하면 됩니다. 또한 엔드포인트 함수는 데이터베이스 쿼리 방법을 알 필요가 없어졌습니다. 이것이 바로 관심사의 분리입니다.
여기서 주목할 점은 user_repo: UserRepository = Depends() 표현입니다. 클래스 이름만 타입 어노테이션으로 제공하고 있는데, 이것은 FastAPI의 또 다른 강력한 기능입니다. FastAPI는 Depends()가 인자 없이 사용될 때 파라미터의 타입 어노테이션을 검사하고, 해당 타입이 클래스라면 그 클래스의 인스턴스를 자동으로 생성하여 주입합니다.
이 메커니즘은 매우 흥미롭습니다. 일반적인 프로그래밍에서 타입 어노테이션은 정적 타입 검사나 IDE의 코드 완성 기능을 위한 힌트로만 사용되지만, FastAPI는 이를 실행 시간에도 활용합니다. user_repo: UserRepository = Depends()라고 작성하면 FastAPI는 다음과 같은 과정을 거칩니다:
1.
파라미터 타입이 UserRepository임을 확인합니다.
2.
Depends()가 인자 없이 사용되었으므로, FastAPI는 타입으로 지정된 UserRepository 클래스를 의존성 클래스로 사용합니다.
3.
UserRepository 클래스의 인스턴스를 생성합니다.
4.
이 과정에서 UserRepository의 생성자에 필요한 인자들도 의존성 주입 시스템을 통해 해결합니다.
이는 user_repo = Depends(UserRepository)와 동일한 효과를 가지지만, 코드가 더 간결하고 타입 힌트를 통해 IDE의 지원을 받을 수 있다는 장점이 있습니다. 이러한 방식은 의존성 주입을 타입 시스템과 결합하여 코드의 가독성과 유지보수성을 높이는 FastAPI의 철학을 잘 보여줍니다.
이제 FastAPI의 의존성 체인(dependency chain)이 어떻게 작동하는지 살펴보겠습니다. 의존성 체인은 하나의 의존성이 다른 의존성에 의존하는 구조를 말합니다. FastAPI는 이러한 의존성 체인을 자동으로 해결해 주며, 이것이 바로 Depends()의 또 다른 강력한 기능입니다.
의존성 체인의 작동 방식을 이해하기 위해 앞서 설명한 UserRepository 클래스의 생성 과정을 더 자세히 살펴보겠습니다:
class BaseRepository:
def __init__(self, db: Session = Depends(get_db)):
self.db = db
class UserRepository(BaseRepository):
def get_by_email(self, email: str):
# 구현...
@router.post("/login/")
def login(credentials: LoginCredentials, user_repo: UserRepository = Depends()):
# 구현...
Python
복사
login 엔드포인트가 호출될 때, FastAPI는 다음과 같은 순서로 의존성 체인을 해결합니다:
1.
user_repo: UserRepository = Depends()를 처리하기 위해 UserRepository 클래스의 인스턴스를 생성해야 함을 파악합니다.
2.
UserRepository는 BaseRepository를 상속하므로 BaseRepository의 생성자가 호출됩니다.
3.
BaseRepository 생성자는 db: Session = Depends(get_db)를 필요로 합니다.
4.
FastAPI는 get_db 함수를 호출하여 데이터베이스 세션을 가져옵니다.
5.
가져온 데이터베이스 세션으로 BaseRepository 인스턴스가 생성됩니다.
6.
이를 기반으로 UserRepository 인스턴스가 완성됩니다.
7.
최종적으로 완성된 UserRepository 인스턴스가 login 함수의 user_repo 파라미터에 주입됩니다.
이 모든 과정이 FastAPI에 의해 자동으로 처리되므로, 개발자는 각 컴포넌트를 독립적으로 정의하고 필요한 의존성만 선언하면 됩니다. 이러한 체인화된 의존성 구조는 코드의 모듈성과 재사용성을 크게 향상시킵니다.
더 복잡한 의존성 체인도 가능합니다. 예를 들어, 설정, 데이터베이스 URL, 엔진, 세션 팩토리, 세션, 리포지토리로 이어지는 체인을 구성할 수 있습니다:
def get_settings():
return Settings() # 환경 설정 객체 반환
def get_db_url(settings = Depends(get_settings)):
return settings.DATABASE_URL # 설정에서 DB URL 가져오기
def get_engine(db_url: str = Depends(get_db_url)):
return create_engine(db_url) # DB 엔진 생성
def get_session_factory(engine = Depends(get_engine)):
return sessionmaker(bind=engine) # 세션 팩토리 생성
def get_db(session_factory = Depends(get_session_factory)):
session = session_factory() # 세션 인스턴스 생성
try:
yield session
finally:
session.close()
class BaseRepository:
def __init__(self, db: Session = Depends(get_db)):
self.db = db
class UserRepository(BaseRepository):
# 구현...
@router.post("/login/")
def login(credentials: LoginCredentials, user_repo: UserRepository = Depends()):
# 구현...
Python
복사
login 엔드포인트가 호출될 때, FastAPI는 get_settings → get_db_url → get_engine → get_session_factory → get_db → BaseRepository → UserRepository로 이어지는 전체 의존성 체인을 해결합니다. 이 모든 것이 자동으로 처리되므로, 개발자는 이러한 복잡성을 신경 쓰지 않고 모듈화된 코드를 작성할 수 있습니다.
특히 테스트 시에 이러한 의존성 체인의 어느 지점이든 오버라이드할 수 있다는 점이 중요합니다. 테스트 DB와 프로덕션 DB를 분리하는 경우, 체인의 가장 상위에 있는 get_settings만 오버라이드하여 테스트용 설정을 제공할 수 있습니다:
def get_test_settings():
return TestSettings() # 테스트용 설정 반환
app.dependency_overrides[get_settings] = get_test_settings
Python
복사
이 간단한 오버라이드로 인해 전체 의존성 체인이 테스트 데이터베이스를 사용하게 됩니다. 애플리케이션 코드는 전혀 변경하지 않았지만, 실행 환경에 따라 다른 데이터베이스를 사용하도록 동적으로 구성되었습니다.
from
1.
•
앞의 메모는 depends의 yield 결합 시의 동작에 대해 간단히 설명한다. 이 글은 앞의 메모에서 설명한 내용이 실제로 워크플로에서 어떻게 결합되는지를 주로 설명하는데, 완결된 글을 만들고자 하는 과정에서 앞의 글과 겹치는 내용이 많이 작성되었다.
reference