FastAPI 데이터베이스부터 배포까지
Posted by Albert 12Day 23Hour 3Min 44Sec ago [2025-12-03]
1. 데이터베이스 연동 - SQLAlchemy
1-1. 설치
pip install sqlalchemy databases aiosqlite
1-2. 데이터베이스 설정
database.py 파일을 생성합니다.
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
' SQLite 데이터베이스 URL (실제 환경에서는 PostgreSQL, MySQL 등 사용)
SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
' 데이터베이스 엔진 생성
engine = create_engine(
SQLALCHEMY_DATABASE_URL,
connect_args={"check_same_thread": False} ' SQLite용 설정
)
' 세션 로컬 클래스 생성
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
' Base 클래스 생성 (모델들이 상속받을 기본 클래스)
Base = declarative_base()
' 데이터베이스 세션 의존성
def get_db():
db = SessionLocal()
try:
yield db
finally:
db.close()
1-3. 모델 정의
models.py 파일을 생성합니다.
from sqlalchemy import Boolean, Column, ForeignKey, Integer, String, DateTime
from sqlalchemy.orm import relationship
from datetime import datetime
from database import Base
class User(Base):
__tablename__ = "users"
id = Column(Integer, primary_key=True, index=True)
email = Column(String, unique=True, index=True)
username = Column(String, unique=True, index=True)
hashed_password = Column(String)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.utcnow)
' 관계 설정
posts = relationship("Post", back_populates="owner")
class Post(Base):
__tablename__ = "posts"
id = Column(Integer, primary_key=True, index=True)
title = Column(String, index=True)
content = Column(String)
published = Column(Boolean, default=False)
created_at = Column(DateTime, default=datetime.utcnow)
owner_id = Column(Integer, ForeignKey("users.id"))
' 관계 설정
owner = relationship("User", back_populates="posts")
1-4. Pydantic 스키마
schemas.py 파일을 생성합니다.
from pydantic import BaseModel, EmailStr
from datetime import datetime
from typing import List, Optional
' Post 스키마
class PostBase(BaseModel):
title: str
content: str
published: bool = False
class PostCreate(PostBase):
pass
class Post(PostBase):
id: int
created_at: datetime
owner_id: int
class Config:
from_attributes = True ' ORM 모드 활성화
' User 스키마
class UserBase(BaseModel):
email: EmailStr
username: str
class UserCreate(UserBase):
password: str
class User(UserBase):
id: int
is_active: bool
created_at: datetime
posts: List[Post] = []
class Config:
from_attributes = True
1-5. CRUD 작업
crud.py 파일을 생성합니다.
from sqlalchemy.orm import Session
import models, schemas
from passlib.context import CryptContext
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
' User CRUD
def get_user(db: Session, user_id: int):
return db.query(models.User).filter(models.User.id == user_id).first()
def get_user_by_email(db: Session, email: str):
return db.query(models.User).filter(models.User.email == email).first()
def get_users(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.User).offset(skip).limit(limit).all()
def create_user(db: Session, user: schemas.UserCreate):
hashed_password = pwd_context.hash(user.password)
db_user = models.User(
email=user.email,
username=user.username,
hashed_password=hashed_password
)
db.add(db_user)
db.commit()
db.refresh(db_user)
return db_user
' Post CRUD
def get_posts(db: Session, skip: int = 0, limit: int = 100):
return db.query(models.Post).offset(skip).limit(limit).all()
def create_user_post(db: Session, post: schemas.PostCreate, user_id: int):
db_post = models.Post(**post.dict(), owner_id=user_id)
db.add(db_post)
db.commit()
db.refresh(db_post)
return db_post
def delete_post(db: Session, post_id: int):
db_post = db.query(models.Post).filter(models.Post.id == post_id).first()
if db_post:
db.delete(db_post)
db.commit()
return True
return False
1-6. API 엔드포인트
main.py에 통합합니다.
from fastapi import FastAPI, Depends, HTTPException
from sqlalchemy.orm import Session
from typing import List
import models, schemas, crud
from database import engine, get_db
' 데이터베이스 테이블 생성
models.Base.metadata.create_all(bind=engine)
app = FastAPI(title="Blog API")
' 사용자 생성
@app.post("/users/", response_model=schemas.User)
def create_user(user: schemas.UserCreate, db: Session = Depends(get_db)):
db_user = crud.get_user_by_email(db, email=user.email)
if db_user:
raise HTTPException(status_code=400, detail="Email already registered")
return crud.create_user(db=db, user=user)
' 사용자 목록 조회
@app.get("/users/", response_model=List[schemas.User])
def read_users(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
users = crud.get_users(db, skip=skip, limit=limit)
return users
' 특정 사용자 조회
@app.get("/users/{user_id}", response_model=schemas.User)
def read_user(user_id: int, db: Session = Depends(get_db)):
db_user = crud.get_user(db, user_id=user_id)
if db_user is None:
raise HTTPException(status_code=404, detail="User not found")
return db_user
' 게시글 생성
@app.post("/users/{user_id}/posts/", response_model=schemas.Post)
def create_post_for_user(
user_id: int,
post: schemas.PostCreate,
db: Session = Depends(get_db)
):
return crud.create_user_post(db=db, post=post, user_id=user_id)
' 게시글 목록 조회
@app.get("/posts/", response_model=List[schemas.Post])
def read_posts(skip: int = 0, limit: int = 100, db: Session = Depends(get_db)):
posts = crud.get_posts(db, skip=skip, limit=limit)
return posts
2. 인증/권한 - JWT
2-1. 설치
pip install python-jose[cryptography] passlib[bcrypt] python-multipart
2-2. 인증 설정
auth.py 파일을 생성합니다.
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from database import get_db
import models, schemas
' 설정
SECRET_KEY = "your-secret-key-keep-it-secret" ' 실제로는 환경변수 사용
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")
' 비밀번호 검증
def verify_password(plain_password, hashed_password):
return pwd_context.verify(plain_password, hashed_password)
def get_password_hash(password):
return pwd_context.hash(password)
' 사용자 인증
def authenticate_user(db: Session, email: str, password: str):
user = db.query(models.User).filter(models.User.email == email).first()
if not user:
return False
if not verify_password(password, user.hashed_password):
return False
return user
' JWT 토큰 생성
def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
to_encode = data.copy()
if expires_delta:
expire = datetime.utcnow() + expires_delta
else:
expire = datetime.utcnow() + timedelta(minutes=15)
to_encode.update({"exp": expire})
encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
return encoded_jwt
' 현재 사용자 가져오기
async def get_current_user(
token: str = Depends(oauth2_scheme),
db: Session = Depends(get_db)
):
credentials_exception = HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Could not validate credentials",
headers={"WWW-Authenticate": "Bearer"},
)
try:
payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
email: str = payload.get("sub")
if email is None:
raise credentials_exception
except JWTError:
raise credentials_exception
user = db.query(models.User).filter(models.User.email == email).first()
if user is None:
raise credentials_exception
return user
' 활성 사용자 확인
async def get_current_active_user(
current_user: models.User = Depends(get_current_user)
):
if not current_user.is_active:
raise HTTPException(status_code=400, detail="Inactive user")
return current_user
2-3. 로그인 엔드포인트
main.py에 추가합니다.
from fastapi.security import OAuth2PasswordRequestForm
from auth import (
authenticate_user,
create_access_token,
get_current_active_user,
ACCESS_TOKEN_EXPIRE_MINUTES
)
' 토큰 스키마
class Token(BaseModel):
access_token: str
token_type: str
' 로그인
@app.post("/token", response_model=Token)
async def login(
form_data: OAuth2PasswordRequestForm = Depends(),
db: Session = Depends(get_db)
):
user = authenticate_user(db, form_data.username, form_data.password)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="Incorrect email or password",
headers={"WWW-Authenticate": "Bearer"},
)
access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
access_token = create_access_token(
data={"sub": user.email}, expires_delta=access_token_expires
)
return {"access_token": access_token, "token_type": "bearer"}
' 보호된 엔드포인트 예제
@app.get("/users/me", response_model=schemas.User)
async def read_users_me(current_user: models.User = Depends(get_current_active_user)):
return current_user
' 인증이 필요한 게시글 생성
@app.post("/posts/", response_model=schemas.Post)
def create_post(
post: schemas.PostCreate,
db: Session = Depends(get_db),
current_user: models.User = Depends(get_current_active_user)
):
return crud.create_user_post(db=db, post=post, user_id=current_user.id)
2-4. 권한 기반 접근 제어 (RBAC)
from enum import Enum
class Role(str, Enum):
ADMIN = "admin"
USER = "user"
GUEST = "guest"
' models.py에 추가
class User(Base):
' ... 기존 컬럼들
role = Column(String, default=Role.USER)
' 권한 체크 함수
def check_permission(required_role: Role):
async def permission_checker(
current_user: models.User = Depends(get_current_active_user)
):
if current_user.role != required_role and current_user.role != Role.ADMIN:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail="Not enough permissions"
)
return current_user
return permission_checker
' 관리자만 접근 가능한 엔드포인트
@app.delete("/posts/{post_id}")
def delete_post(
post_id: int,
db: Session = Depends(get_db),
current_user: models.User = Depends(check_permission(Role.ADMIN))
):
if not crud.delete_post(db, post_id):
raise HTTPException(status_code=404, detail="Post not found")
return {"message": "Post deleted"}
3. 백그라운드 태스크
3-1. 간단한 백그라운드 태스크
from fastapi import BackgroundTasks
import time
def write_log(message: str):
time.sleep(5) ' 시간이 걸리는 작업 시뮬레이션
with open("log.txt", "a") as f:
f.write(f"{datetime.now()}: {message}\n")
@app.post("/send-notification/")
async def send_notification(
email: str,
background_tasks: BackgroundTasks
):
' 즉시 응답 반환
background_tasks.add_task(write_log, f"Notification sent to {email}")
return {"message": "Notification sent in the background"}
3-2. 이메일 발송 백그라운드 태스크
pip install fastapi-mail
from fastapi_mail import FastMail, MessageSchema, ConnectionConfig
from pydantic import EmailStr
' 이메일 설정
conf = ConnectionConfig(
MAIL_USERNAME="your-email@example.com",
MAIL_PASSWORD="your-password",
MAIL_FROM="your-email@example.com",
MAIL_PORT=587,
MAIL_SERVER="smtp.gmail.com",
MAIL_STARTTLS=True,
MAIL_SSL_TLS=False
)
async def send_welcome_email(email: EmailStr, username: str):
message = MessageSchema(
subject="Welcome to Our Service",
recipients=[email],
body=f"Hello {username}, welcome to our service!",
subtype="html"
)
fm = FastMail(conf)
await fm.send_message(message)
@app.post("/register/")
async def register_user(
user: schemas.UserCreate,
background_tasks: BackgroundTasks,
db: Session = Depends(get_db)
):
db_user = crud.create_user(db=db, user=user)
' 백그라운드에서 환영 이메일 발송
background_tasks.add_task(send_welcome_email, db_user.email, db_user.username)
return db_user
3-3. Celery를 사용한 고급 백그라운드 작업
pip install celery redis
celery_worker.py 파일을 생성합니다.
from celery import Celery
celery_app = Celery(
"tasks",
broker="redis://localhost:6379/0",
backend="redis://localhost:6379/0"
)
@celery_app.task
def process_data(data: dict):
' 복잡한 데이터 처리 작업
time.sleep(10)
return {"status": "processed", "data": data}
@celery_app.task
def generate_report(user_id: int):
' 리포트 생성 작업
time.sleep(30)
return {"report": "generated", "user_id": user_id}
main.py에서 사용:
from celery_worker import process_data, generate_report
@app.post("/process/")
async def process_request(data: dict):
task = process_data.delay(data)
return {"task_id": task.id, "status": "processing"}
@app.get("/task/{task_id}")
async def get_task_status(task_id: str):
task = celery_app.AsyncResult(task_id)
return {
"task_id": task_id,
"status": task.status,
"result": task.result if task.ready() else None
}
Celery worker 실행:
celery -A celery_worker worker --loglevel=info
4. 테스트 작성
4-1. 설치
pip install pytest httpx pytest-asyncio
4-2. 테스트 설정
test_main.py 파일을 생성합니다.
from fastapi.testclient import TestClient
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from main import app
from database import Base, get_db
' 테스트용 데이터베이스
SQLALCHEMY_DATABASE_URL = "sqlite:///./test_db.db"
engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
' 테이블 생성
Base.metadata.create_all(bind=engine)
def override_get_db():
try:
db = TestingSessionLocal()
yield db
finally:
db.close()
app.dependency_overrides[get_db] = override_get_db
client = TestClient(app)
4-3. 기본 테스트
def test_read_main():
response = client.get("/")
assert response.status_code == 200
def test_create_user():
response = client.post(
"/users/",
json={
"email": "test@example.com",
"username": "testuser",
"password": "testpass123"
}
)
assert response.status_code == 200
data = response.json()
assert data["email"] == "test@example.com"
assert "id" in data
def test_create_duplicate_user():
' 첫 번째 사용자 생성
client.post(
"/users/",
json={
"email": "duplicate@example.com",
"username": "duplicate",
"password": "pass123"
}
)
' 중복 사용자 생성 시도
response = client.post(
"/users/",
json={
"email": "duplicate@example.com",
"username": "duplicate2",
"password": "pass123"
}
)
assert response.status_code == 400
assert response.json()["detail"] == "Email already registered"
def test_read_users():
response = client.get("/users/")
assert response.status_code == 200
assert isinstance(response.json(), list)
4-4. 인증 테스트
def test_login():
' 먼저 사용자 생성
client.post(
"/users/",
json={
"email": "login@example.com",
"username": "loginuser",
"password": "loginpass123"
}
)
' 로그인
response = client.post(
"/token",
data={
"username": "login@example.com",
"password": "loginpass123"
}
)
assert response.status_code == 200
assert "access_token" in response.json()
assert response.json()["token_type"] == "bearer"
def test_protected_endpoint():
' 토큰 없이 접근
response = client.get("/users/me")
assert response.status_code == 401
' 토큰 획득
login_response = client.post(
"/token",
data={
"username": "login@example.com",
"password": "loginpass123"
}
)
token = login_response.json()["access_token"]
' 토큰으로 접근
response = client.get(
"/users/me",
headers={"Authorization": f"Bearer {token}"}
)
assert response.status_code == 200
4-5. Pytest Fixtures 사용
import pytest
@pytest.fixture
def test_user():
response = client.post(
"/users/",
json={
"email": "fixture@example.com",
"username": "fixtureuser",
"password": "pass123"
}
)
return response.json()
@pytest.fixture
def auth_token(test_user):
response = client.post(
"/token",
data={
"username": test_user["email"],
"password": "pass123"
}
)
return response.json()["access_token"]
def test_create_post_authenticated(auth_token):
response = client.post(
"/posts/",
json={
"title": "Test Post",
"content": "Test content",
"published": True
},
headers={"Authorization": f"Bearer {auth_token}"}
)
assert response.status_code == 200
assert response.json()["title"] == "Test Post"
테스트 실행:
pytest pytest -v ' 상세 출력 pytest --cov=. ' 코드 커버리지
5. 배포 - Docker
5-1. Dockerfile 작성
프로젝트 루트에 Dockerfile을 생성합니다.
' Python 3.11 이미지 사용
FROM python:3.11-slim
' 작업 디렉토리 설정
WORKDIR /app
' 시스템 패키지 업데이트
RUN apt-get update && apt-get install -y \
gcc \
&& rm -rf /var/lib/apt/lists/*
' requirements.txt 복사 및 설치
COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt
' 애플리케이션 코드 복사
COPY . .
' 포트 노출
EXPOSE 8000
' 애플리케이션 실행
CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]
5-2. requirements.txt 생성
fastapi==0.104.1 uvicorn[standard]==0.24.0 sqlalchemy==2.0.23 pydantic[email]==2.5.0 python-jose[cryptography]==3.3.0 passlib[bcrypt]==1.7.4 python-multipart==0.0.6
5-3. docker-compose.yml 작성
version: '3.8'
services:
web:
build: .
ports:
- "8000:8000"
volumes:
- .:/app
environment:
- DATABASE_URL=postgresql://user:password@db:5432/dbname
depends_on:
- db
command: uvicorn main:app --host 0.0.0.0 --port 8000 --reload
db:
image: postgres:15
environment:
- POSTGRES_USER=user
- POSTGRES_PASSWORD=password
- POSTGRES_DB=dbname
volumes:
- postgres_data:/var/lib/postgresql/data
ports:
- "5432:5432"
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
postgres_data:
```
''' 5-4. .dockerignore 작성
```
__pycache__
*.pyc
*.pyo
*.pyd
.Python
env/
venv/
.venv
.git
.gitignore
.pytest_cache
.coverage
*.db
*.log
5-5. Docker 명령어
' 이미지 빌드 docker build -t fastapi-app . ' 컨테이너 실행 docker run -d -p 8000:8000 fastapi-app ' docker-compose로 실행 docker-compose up -d ' 로그 확인 docker-compose logs -f ' 중지 docker-compose down ' 볼륨까지 삭제 docker-compose down -v ``` --- '' 6. 클라우드 배포 ''' 6-1. Heroku 배포 `Procfile` 생성: ``` web: uvicorn main:app --host 0.0.0.0 --port $PORT ``` `runtime.txt` 생성: ``` python-3.11.0
배포 명령:
' Heroku CLI 설치 후 heroku login heroku create your-app-name git push heroku main heroku ps:scale web=1 heroku open
6-2. AWS EC2 배포
' EC2 인스턴스 접속 후
sudo apt update
sudo apt install python3-pip python3-venv nginx
' 프로젝트 클론
git clone your-repo-url
cd your-project
' 가상환경 설정
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
' Gunicorn 설치
pip install gunicorn
' systemd 서비스 파일 생성 (/etc/systemd/system/fastapi.service)
[Unit]
Description=FastAPI application
After=network.target
[Service]
User=ubuntu
WorkingDirectory=/home/ubuntu/your-project
Environment="PATH=/home/ubuntu/your-project/venv/bin"
ExecStart=/home/ubuntu/your-project/venv/bin/gunicorn -w 4 -k uvicorn.workers.UvicornWorker main:app --bind 0.0.0.0:8000
[Install]
WantedBy=multi-user.target
' 서비스 시작
sudo systemctl start fastapi
sudo systemctl enable fastapi
' Nginx 설정 (/etc/nginx/sites-available/fastapi)
server {
listen 80;
server_name your-domain.com;
location / {
proxy_pass http://127.0.0.1:8000;
proxy_set_header Host $host;
proxy_set_header X-Real-IP $remote_addr;
}
}
' Nginx 활성화
sudo ln -s /etc/nginx/sites-available/fastapi /etc/nginx/sites-enabled/
sudo nginx -t
sudo systemctl restart nginx
```
''' 6-3. 환경변수 관리
`.env` 파일 생성:
```
DATABASE_URL=postgresql://user:password@localhost/dbname
SECRET_KEY=your-secret-key-here
ALGORITHM=HS256
ACCESS_TOKEN_EXPIRE_MINUTES=30
config.py 생성:
from pydantic_settings import BaseSettings
class Settings(BaseSettings):
database_url: str
secret_key: str
algorithm: str = "HS256"
access_token_expire_minutes: int = 30
class Config:
env_file = ".env"
settings = Settings()
사용:
from config import settings SECRET_KEY = settings.secret_key
6-4. HTTPS 설정 (Let's Encrypt)
' Certbot 설치 sudo apt install certbot python3-certbot-nginx ' SSL 인증서 발급 sudo certbot --nginx -d your-domain.com ' 자동 갱신 설정 sudo certbot renew --dry-run ``` --- '' 프로젝트 구조 예시 ``` fastapi-project/ │ ├── app/ │ ├── __init__.py │ ├── main.py │ ├── config.py │ ├── database.py │ │ │ ├── models/ │ │ ├── __init__.py │ │ └── user.py │ │ │ ├── schemas/ │ │ ├── __init__.py │ │ └── user.py │ │ │ ├── crud/ │ │ ├── __init__.py │ │ └── user.py │ │ │ ├── api/ │ │ ├── __init__.py │ │ ├── deps.py │ │ └── endpoints/ │ │ ├── __init__.py │ │ ├── users.py │ │ └── posts.py │ │ │ └── core/ │ ├── __init__.py │ ├── security.py │ └── config.py │ ├── tests/ │ ├── __init__.py │ ├── test_users.py │ └── test_auth.py │ ├── .env ├── .gitignore ├── Dockerfile ├── docker-compose.yml ├── requirements.txt └── README.md