FastAPI로 고성능 REST API 만들기 - Flask 대안 탐색

FastAPI로 고성능 REST API 만들기 - Flask 대안 탐색

 

FastAPI는 파이썬 웹 프레임워크 생태계에서 혁신적인 변화를 가져온 모던 프레임워크입니다.

Flask보다 더 빠르고 직관적인 개발 경험을 제공하는 FastAPI는 REST API 개발에 최적화된 특징들을 갖추고 있습니다.

이 글에서는 FastAPI가 어떻게 기존 Flask의 한계를 극복하고, 어떤 장점들을 제공하는지 실제 코드 예제와 함께 살펴보겠습니다.

---

FastAPI란 무엇인가?

FastAPI는 Sebastian Ramirez가 개발한 현대적인 파이썬 웹 프레임워크입니다.

Python 3.6+의 타입 힌트를 기반으로 구축되어 있으며, 자동 API 문서 생성과 데이터 검증 기능을 내장하고 있습니다.

FastAPI 아키텍처 다이어그램

FastAPI의 핵심 특징은 다음과 같습니다:

• 고성능: Node.js나 Go와 비슷한 성능을 제공합니다.
• 타입 안전성: Python 타입 힌트를 통한 자동 검증과 IDE 지원을 받을 수 있습니다.
• 자동 문서화: OpenAPI와 JSON Schema를 기반으로 한 자동 API 문서 생성이 가능합니다.
• 비동기 지원: async/await를 네이티브로 지원하여 동시성 처리가 뛰어납니다.

from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float
    is_offer: bool = None

@app.get("/")
def read_root():
    return {"Hello": "World"}

@app.get("/items/{item_id}")
def read_item(item_id: int, q: str = None):
    return {"item_id": item_id, "q": q}

@app.post("/items/")
def create_item(item: Item):
    return item

FastAPI 공식 문서에서 더 자세한 정보를 확인할 수 있습니다.

---

Flask vs FastAPI 성능 비교

성능은 웹 API 선택에서 가장 중요한 고려사항 중 하나입니다.

FastAPI는 Starlette ASGI 프레임워크를 기반으로 구축되어 Flask보다 현저히 빠른 성능을 보여줍니다.

벤치마크 결과

실제 벤치마크 테스트에서 FastAPI는 Flask보다 약 2-3배 빠른 응답 속도를 보여줍니다.

특히 동시 요청 처리에서 그 차이가 더욱 두드러집니다.

Flask vs FastAPI 성능 비교 차트

동시성 처리 비교

Flask는 기본적으로 동기 방식으로 작동하여 하나의 요청이 완료될 때까지 다른 요청들이 대기해야 합니다.

반면 FastAPI는 비동기 처리를 기본으로 지원하여 여러 요청을 동시에 처리할 수 있습니다.

# Flask - 동기 방식
from flask import Flask
import time

app = Flask(__name__)

@app.route('/slow-task')
def slow_task():
    time.sleep(5)  # 5초 대기
    return {"message": "Task completed"}

# FastAPI - 비동기 방식
from fastapi import FastAPI
import asyncio

app = FastAPI()

@app.get('/slow-task')
async def slow_task():
    await asyncio.sleep(5)  # 5초 비동기 대기
    return {"message": "Task completed"}

Starlette 공식 문서에서 ASGI의 장점에 대해 더 자세히 알아볼 수 있습니다.

---

자동 API 문서 생성과 데이터 검증

FastAPI의 가장 강력한 기능 중 하나는 자동 API 문서 생성입니다.

코드를 작성하면 자동으로 OpenAPI 스펙에 맞는 문서가 생성되며,

Swagger UI와 ReDoc을 통해 인터랙티브한 문서를 제공받을 수 있습니다.

Pydantic을 통한 데이터 검증

FastAPI는 Pydantic 모델을 사용하여 요청과 응답 데이터를 자동으로 검증합니다.

이는 런타임 에러를 크게 줄여주고 더욱 안전한 API를 만들 수 있게 해줍니다.

from fastapi import FastAPI, HTTPException
from pydantic import BaseModel, validator
from typing import Optional
from datetime import datetime

app = FastAPI(
    title="사용자 관리 API",
    description="FastAPI를 사용한 사용자 관리 시스템",
    version="1.0.0"
)

class UserCreate(BaseModel):
    username: str
    email: str
    age: int
    is_active: bool = True

    @validator('age')
    def validate_age(cls, v):
        if v < 0 or v > 150:
            raise ValueError('나이는 0과 150 사이여야 합니다')
        return v

    @validator('email')
    def validate_email(cls, v):
        if '@' not in v:
            raise ValueError('유효한 이메일 주소를 입력해주세요')
        return v

class UserResponse(BaseModel):
    id: int
    username: str
    email: str
    age: int
    is_active: bool
    created_at: datetime

@app.post("/users/", response_model=UserResponse)
async def create_user(user: UserCreate):
    # 실제로는 데이터베이스에 저장
    new_user = UserResponse(
        id=1,
        username=user.username,
        email=user.email,
        age=user.age,
        is_active=user.is_active,
        created_at=datetime.now()
    )
    return new_user

 

FastAPI Swagger UI 스크린샷

 

Pydantic 공식 문서에서 더 많은 검증 옵션을 확인할 수 있습니다.

자동 문서화의 장점

FastAPI를 사용하면 별도의 문서 작성 없이도 다음과 같은 기능들을 자동으로 얻을 수 있습니다:

• Swagger UI: /docs 엔드포인트에서 인터랙티브한 API 문서를 확인할 수 있습니다.
• ReDoc: /redoc 엔드포인트에서 더욱 세련된 문서 형태로 API를 확인할 수 있습니다.
• OpenAPI 스키마: JSON 형태의 API 스키마를 /openapi.json에서 확인할 수 있습니다.

---

실전 FastAPI 프로젝트 구조

반응형

대규모 FastAPI 애플리케이션을 개발할 때는 적절한 프로젝트 구조가 필요합니다.

다음은 실제 프로덕션 환경에서 사용할 수 있는 FastAPI 프로젝트 구조의 예시입니다.

fastapi_project/
├── app/
│   ├── __init__.py
│   ├── main.py
│   ├── core/
│   │   ├── __init__.py
│   │   ├── config.py
│   │   └── security.py
│   ├── api/
│   │   ├── __init__.py
│   │   ├── api_v1/
│   │   │   ├── __init__.py
│   │   │   ├── api.py
│   │   │   └── endpoints/
│   │   │       ├── __init__.py
│   │   │       ├── users.py
│   │   │       └── items.py
│   ├── models/
│   │   ├── __init__.py
│   │   └── user.py
│   ├── schemas/
│   │   ├── __init__.py
│   │   └── user.py
│   └── crud/
│       ├── __init__.py
│       └── crud_user.py
├── requirements.txt
└── README.md

설정 관리

FastAPI에서는 환경 변수를 통한 설정 관리가 매우 중요합니다.

Pydantic의 BaseSettings를 사용하면 타입 안전한 설정 관리가 가능합니다.

# app/core/config.py
from pydantic import BaseSettings
from typing import Optional

class Settings(BaseSettings):
    app_name: str = "FastAPI 애플리케이션"
    debug: bool = False
    database_url: str
    secret_key: str
    access_token_expire_minutes: int = 30

    class Config:
        env_file = ".env"

settings = Settings()

의존성 주입 시스템

FastAPI의 의존성 주입 시스템은 코드의 재사용성과 테스트 용이성을 크게 향상시킵니다.

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import verify_token

security = HTTPBearer()

def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db: Session = Depends(get_db)
):
    token = credentials.credentials
    user = verify_token(token, db)
    if not user:
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="유효하지 않은 토큰입니다"
        )
    return user

@app.get("/protected")
async def protected_route(current_user = Depends(get_current_user)):
    return {"message": f"안녕하세요, {current_user.username}님!"}

FastAPI 의존성 주입 가이드에서 더 자세한 내용을 확인할 수 있습니다.

---

데이터베이스 연동과 ORM

FastAPI는 다양한 데이터베이스와 ORM을 지원합니다.

가장 널리 사용되는 SQLAlchemy와의 연동 방법을 살펴보겠습니다.

SQLAlchemy 설정

# app/core/database.py
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from app.core.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
    finally:
        db.close()

모델 정의

# app/models/user.py
from sqlalchemy import Boolean, Column, Integer, String, DateTime
from sqlalchemy.sql import func
from app.core.database import Base

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String, unique=True, index=True)
    email = Column(String, unique=True, index=True)
    hashed_password = Column(String)
    is_active = Column(Boolean, default=True)
    created_at = Column(DateTime, server_default=func.now())
    updated_at = Column(DateTime, server_default=func.now(), onupdate=func.now())

CRUD 연산

# app/crud/crud_user.py
from sqlalchemy.orm import Session
from app.models.user import User
from app.schemas.user import UserCreate
from passlib.context import CryptContext

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

class CRUDUser:
    def get(self, db: Session, id: int):
        return db.query(User).filter(User.id == id).first()

    def get_by_email(self, db: Session, email: str):
        return db.query(User).filter(User.email == email).first()

    def create(self, db: Session, user_in: UserCreate):
        hashed_password = pwd_context.hash(user_in.password)
        db_user = User(
            username=user_in.username,
            email=user_in.email,
            hashed_password=hashed_password
        )
        db.add(db_user)
        db.commit()
        db.refresh(db_user)
        return db_user

    def get_multi(self, db: Session, skip: int = 0, limit: int = 100):
        return db.query(User).offset(skip).limit(limit).all()

crud_user = CRUDUser()

SQLAlchemy 공식 문서에서 더 많은 ORM 기능을 확인할 수 있습니다.

---

인증과 보안

웹 API에서 보안은 필수적인 요소입니다.

FastAPI는 다양한 보안 기능을 내장하고 있어 안전한 API를 쉽게 구축할 수 있습니다.

JWT 토큰 기반 인증

# app/core/security.py
from datetime import datetime, timedelta
from typing import Optional
from jose import JWTError, jwt
from passlib.context import CryptContext
from app.core.config import settings

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")

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=settings.access_token_expire_minutes)

    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, settings.secret_key, algorithm="HS256")
    return encoded_jwt

def verify_password(plain_password: str, hashed_password: str) -> bool:
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password: str) -> str:
    return pwd_context.hash(password)

로그인 엔드포인트

from fastapi import APIRouter, Depends, HTTPException, status
from fastapi.security import OAuth2PasswordRequestForm
from sqlalchemy.orm import Session
from app.core.database import get_db
from app.core.security import verify_password, create_access_token
from app.crud.crud_user import crud_user

router = APIRouter()

@router.post("/login")
async def login(
    db: Session = Depends(get_db),
    form_data: OAuth2PasswordRequestForm = Depends()
):
    user = crud_user.get_by_email(db, email=form_data.username)
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="이메일 또는 비밀번호가 올바르지 않습니다",
            headers={"WWW-Authenticate": "Bearer"},
        )

    access_token = create_access_token(data={"sub": user.email})
    return {"access_token": access_token, "token_type": "bearer"}

FastAPI 보안 가이드에서 더 다양한 보안 기능을 확인할 수 있습니다.

---

테스팅과 디버깅

FastAPI는 테스트 작성이 매우 간편합니다.

pytest와 TestClient를 사용하여 API 엔드포인트를 쉽게 테스트할 수 있습니다.

단위 테스트 작성

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

client = TestClient(app)

def test_create_user():
    response = client.post(
        "/users/",
        json={
            "username": "testuser",
            "email": "test@example.com",
            "age": 25
        }
    )
    assert response.status_code == 200
    data = response.json()
    assert data["username"] == "testuser"
    assert data["email"] == "test@example.com"
    assert "id" in data

def test_read_user():
    response = client.get("/users/1")
    assert response.status_code == 200
    data = response.json()
    assert data["id"] == 1

def test_invalid_user_data():
    response = client.post(
        "/users/",
        json={
            "username": "testuser",
            "email": "invalid-email",
            "age": -5
        }
    )
    assert response.status_code == 422  # Validation Error

통합 테스트와 테스트 데이터베이스

# tests/conftest.py
import pytest
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from fastapi.testclient import TestClient
from app.main import app
from app.core.database import get_db, Base

SQLALCHEMY_DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(
    SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False}
)
TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

@pytest.fixture
def db():
    Base.metadata.create_all(bind=engine)
    db = TestingSessionLocal()
    try:
        yield db
    finally:
        db.close()
        Base.metadata.drop_all(bind=engine)

@pytest.fixture
def client(db):
    def override_get_db():
        try:
            yield db
        finally:
            db.close()

    app.dependency_overrides[get_db] = override_get_db
    yield TestClient(app)
    app.dependency_overrides.clear()

Pytest 공식 문서에서 더 많은 테스팅 기법을 확인할 수 있습니다.

---

배포와 운영

FastAPI 애플리케이션을 프로덕션 환경에 배포하는 방법을 살펴보겠습니다.

가장 일반적인 방법인 Docker와 Uvicorn을 사용한 배포 방식을 다루겠습니다.

Docker를 사용한 배포

# Dockerfile
FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY ./app /app

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000"]

# docker-compose.yml
version: '3.8'

services:
  web:
    build: .
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:password@db:5432/dbname
      - SECRET_KEY=your-secret-key
    depends_on:
      - db

  db:
    image: postgres:13
    environment:
      - POSTGRES_USER=user
      - POSTGRES_PASSWORD=password
      - POSTGRES_DB=dbname
    volumes:
      - postgres_data:/var/lib/postgresql/data

volumes:
  postgres_data:

Gunicorn과 Uvicorn Workers

프로덕션 환경에서는 Gunicorn과 Uvicorn worker를 함께 사용하는 것이 권장됩니다.

# 설치
pip install gunicorn[uvicorn]

# 실행
gunicorn app.main:app -w 4 -k uvicorn.workers.UvicornWorker --bind 0.0.0.0:8000

# gunicorn.conf.py
bind = "0.0.0.0:8000"
workers = 4
worker_class = "uvicorn.workers.UvicornWorker"
worker_connections = 1000
max_requests = 1000
max_requests_jitter = 100
timeout = 30
keepalive = 2

FastAPI 배포 가이드에서 더 자세한 배포 방법을 확인할 수 있습니다.

---

모니터링과 로깅

프로덕션 환경에서는 적절한 모니터링과 로깅이 필수적입니다.

FastAPI 애플리케이션의 모니터링과 로깅 설정 방법을 알아보겠습니다.

구조화된 로깅

# app/core/logging.py
import logging
import sys
from typing import Optional
from loguru import logger
from app.core.config import settings

class InterceptHandler(logging.Handler):
    def emit(self, record):
        logger_opt = logger.opt(depth=6, exception=record.exc_info)
        logger_opt.log(record.levelname, record.getMessage())

def setup_logging():
    logging.root.handlers = [InterceptHandler()]
    logging.root.setLevel(logging.INFO)

    for name in logging.root.manager.loggerDict.keys():
        logging.getLogger(name).handlers = []
        logging.getLogger(name).propagate = True

    logger.configure(
        handlers=[
            {
                "sink": sys.stdout,
                "serialize": settings.debug,
                "format": "{time} | {level} | {message}",
            }
        ]
    )

미들웨어를 통한 요청 로깅

from fastapi import Request, Response
from starlette.middleware.base import BaseHTTPMiddleware
import time
from loguru import logger

class LoggingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request: Request, call_next):
        start_time = time.time()

        # 요청 로깅
        logger.info(f"요청 시작: {request.method} {request.url}")

        response = await call_next(request)

        # 응답 로깅
        process_time = time.time() - start_time
        logger.info(
            f"요청 완료: {request.method} {request.url} - "
            f"상태코드: {response.status_code} - "
            f"처리시간: {process_time:.4f}초"
        )

        response.headers["X-Process-Time"] = str(process_time)
        return response

# main.py에서 미들웨어 추가
app.add_middleware(LoggingMiddleware)

Loguru 공식 문서에서 더 많은 로깅 기능을 확인할 수 있습니다.

---

FastAPI vs Flask 마이그레이션 가이드

기존 Flask 애플리케이션을 FastAPI로 마이그레이션하는 과정은 생각보다 간단합니다.

주요 차이점과 마이그레이션 전략을 살펴보겠습니다.

라우팅 방식 비교

# Flask 방식
from flask import Flask, request, jsonify

app = Flask(__name__)

@app.route('/users', methods=['POST'])
def create_user():
    data = request.get_json()
    # 데이터 검증 필요
    user = create_user_service(data)
    return jsonify(user)

# FastAPI 방식
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class UserCreate(BaseModel):
    name: str
    email: str

@app.post('/users')
async def create_user(user: UserCreate):
    # 자동 데이터 검증
    user_data = create_user_service(user.dict())
    return user_data

점진적 마이그레이션 전략

1. 새로운 엔드포인트부터 FastAPI로 개발: 기존 Flask 애플리케이션과 병행하여 새로운 기능을 FastAPI로 구현합니다.
2. 마이크로서비스 분리: 특정 기능을 FastAPI로 분리하여 독립적인 서비스로 운영합니다.
3. 단계별 전환: 엔드포인트별로 순차적으로 FastAPI로 전환합니다.

# 하이브리드 접근법 예시
from fastapi import FastAPI
from flask import Flask
from werkzeug.middleware.dispatcher import DispatcherMiddleware

# FastAPI 앱
fastapi_app = FastAPI()

@fastapi_app.get("/api/v2/users")
async def get_users_v2():
    return {"version": "v2", "users": []}

# Flask 앱 (기존)
flask_app = Flask(__name__)

@flask_app.route("/api/v1/users")
def get_users_v1():
    return {"version": "v1", "users": []}

# 두 앱을 결합
app = DispatcherMiddleware(flask_app, {
    '/api/v2': fastapi_app
})

---

결론

FastAPI는 현대적인 파이썬 웹 개발의 새로운 표준으로 자리잡고 있습니다.

Flask보다 뛰어난 성능과 개발자 경험을 제공하며, 자동 문서화와 타입 안전성을 통해 더욱 안정적인 API를 구축할 수 있게 해줍니다.

 

특히 다음과 같은 프로젝트에서 FastAPI의 장점이 극대화됩니다:

• 고성능이 요구되는 API 서비스: 비동기 처리를 통한 뛰어난 동시성 성능을 활용할 수 있습니다.
• 마이크로서비스 아키텍처: 경량화되고 독립적인 서비스 구축에 적합합니다.
• 데이터 집약적 애플리케이션: Pydantic을 통한 강력한 데이터 검증 기능을 활용할 수 있습니다.
• API 우선 개발: 자동 문서화를 통해 프론트엔드와의 협업을 원활하게 할 수 있습니다.

FastAPI는 학습 곡선이 완만하면서도 강력한 기능을 제공하여, Flask 개발자들이 쉽게 전환할 수 있는 프레임워크입니다.

특히 타입 힌트에 익숙한 개발자라면 더욱 빠르게 적응할 수 있을 것입니다.

앞으로 파이썬 웹 개발을 고려하고 있다면 FastAPI를 적극 고려해보시기 바랍니다.

현재 개발 중인 프로젝트가 있다면 단계적 마이그레이션을 통해 FastAPI의 장점을 경험해보세요.

FastAPI 커뮤니티에서 더 많은 정보와 예제를 확인할 수 있습니다.