CodeDeck

AsyncIO 완벽 마스터

AsyncIO의 핵심 개념과 실전 활용법

Python중급
8시간
4개 항목
학습 진행률0 / 4 (0%)

학습 항목

1. Python
FastAPI|REST|API|개발|완벽|가이드
퀴즈튜토리얼
2. Python
중급
Python 실전 프로젝트 완벽 가이드
퀴즈튜토리얼
3. Python
고급
Python|최신|기능|완벽|가이드
퀴즈튜토리얼
4. Python
고급
RabbitMQ|실무|활용|팁
퀴즈튜토리얼
1 / 4

이미지 로딩 중...

FastAPI REST API 개발 완벽 가이드 - 슬라이드 1/11

FastAPI REST API 개발 완벽 가이드

FastAPI를 활용한 REST API 개발의 핵심 개념부터 실무 활용까지 다룹니다. 초급 개발자도 쉽게 따라할 수 있는 실전 예제와 팁을 제공합니다.

목차

  1. FastAPI 기본 구조와 라우팅
  2. 경로 매개변수와 쿼리 매개변수
  3. Pydantic 모델과 데이터 검증
  4. 비동기 처리와 async/await
  5. 의존성 주입 시스템
  6. 에러 핸들링과 HTTP 예외
  7. 미들웨어와 요청/응답 처리
  8. 데이터베이스 연동 (SQLAlchemy)
  9. 인증과 JWT 토큰
  10. 백그라운드 작업과 Celery 연동

1. FastAPI 기본 구조와 라우팅

시작하며

여러분이 새로운 웹 API를 만들어야 하는데, 어떤 프레임워크를 선택해야 할지 고민되시나요? Flask는 너무 간단하고, Django는 너무 무겁게 느껴지고, Node.js는 파이썬이 아니어서 망설여지는 상황을 경험하신 적 있으실 겁니다.

이런 고민은 많은 개발자들이 겪는 문제입니다. 빠른 개발 속도와 높은 성능, 그리고 자동 문서화까지 원하지만 복잡한 설정은 피하고 싶은 게 개발자의 마음입니다.

바로 이럴 때 필요한 것이 FastAPI입니다. 최소한의 코드로 고성능 API를 만들고, 타입 힌트만으로 자동 검증과 문서화까지 제공받을 수 있습니다.

개요

간단히 말해서, FastAPI는 Python 3.6+ 타입 힌트를 기반으로 하는 현대적이고 빠른 웹 프레임워크입니다. FastAPI가 필요한 이유는 개발 생산성과 런타임 성능을 동시에 잡을 수 있기 때문입니다.

코드를 작성하면 자동으로 API 문서가 생성되고, 타입 체크도 자동으로 이루어집니다. 예를 들어, 사용자 정보를 받는 API를 만들 때 일일이 검증 코드를 작성할 필요가 없습니다.

기존 Flask에서는 @app.route()로 경로를 정의하고 수동으로 데이터 검증을 해야 했다면, FastAPI에서는 타입 힌트만 추가하면 모든 검증이 자동으로 처리됩니다. FastAPI의 핵심 특징은 첫째, Starlette 기반의 비동기 처리로 NodeJS급 성능을 냅니다.

둘째, Pydantic을 통한 자동 데이터 검증이 이루어집니다. 셋째, OpenAPI 기반의 자동 문서화를 제공합니다.

이러한 특징들이 실무에서 API 개발 시간을 절반 이하로 줄여줍니다.

코드 예제

from fastapi import FastAPI
from pydantic import BaseModel

# FastAPI 인스턴스 생성
app = FastAPI()

# 데이터 모델 정의
class User(BaseModel):
    name: str
    age: int
    email: str

# GET 엔드포인트 - 단순 조회
@app.get("/")
async def root():
    return {"message": "FastAPI 서버가 실행 중입니다"}

# POST 엔드포인트 - 데이터 생성
@app.post("/users/")
async def create_user(user: User):
    # Pydantic이 자동으로 검증 수행
    return {"user_id": 1, "name": user.name, "age": user.age}

설명

이것이 하는 일: FastAPI 애플리케이션의 가장 기본적인 구조를 만들고, GET과 POST 두 가지 HTTP 메서드를 처리하는 엔드포인트를 정의합니다. 첫 번째로, FastAPI 인스턴스를 생성하고 Pydantic BaseModel을 상속받은 User 클래스를 정의합니다.

이 User 클래스는 단순한 데이터 클래스가 아니라, 들어오는 요청 데이터를 자동으로 검증하고 변환하는 역할을 합니다. 예를 들어, age 필드에 문자열이 들어오면 자동으로 422 에러를 반환합니다.

그 다음으로, @app.get("/")과 @app.post("/users/") 데코레이터를 사용하여 라우트를 정의합니다. 여기서 async def를 사용하면 비동기 처리가 가능해져, 데이터베이스 쿼리나 외부 API 호출 시 다른 요청을 블로킹하지 않고 처리할 수 있습니다.

이것이 FastAPI의 고성능 비결입니다. 마지막으로, create_user 함수의 파라미터로 user: User를 선언하면, FastAPI가 자동으로 요청 본문을 파싱하고 User 모델에 맞게 검증합니다.

검증에 실패하면 자동으로 상세한 에러 메시지를 반환하므로, 여러분이 별도의 에러 핸들링 코드를 작성할 필요가 없습니다. 여러분이 이 코드를 사용하면 uvicorn main:app --reload 명령어 하나로 서버를 실행하고, http://localhost:8000/docs 에서 자동 생성된 Swagger UI 문서를 확인할 수 있습니다.

실무에서는 API 명세서를 따로 작성할 필요가 없어지고, 프론트엔드 개발자와의 협업이 훨씬 수월해집니다.

실전 팁

💡 서버 실행은 uvicorn main:app --reload --host 0.0.0.0 --port 8000 명령어를 사용하세요. --reload 옵션은 개발 중 코드 변경을 자동 반영합니다

💡 http://localhost:8000/docs에서 Swagger UI, /redoc에서 ReDoc 문서를 자동으로 확인할 수 있습니다. 별도의 문서 작성이 불필요합니다

💡 async def 대신 def를 사용해도 되지만, I/O 작업이 많다면 async를 사용하여 성능을 크게 향상시킬 수 있습니다

💡 Pydantic 모델에 example 값을 추가하면 문서에 예시가 표시되어 API 사용자가 이해하기 쉬워집니다

💡 개발 단계에서는 --reload를 사용하되, 프로덕션 배포 시에는 제거하여 보안과 성능을 확보하세요

2. 경로 매개변수와 쿼리 매개변수

시작하며

여러분이 사용자 ID로 특정 사용자를 조회하거나, 페이지네이션을 구현해야 할 때 어떻게 하시나요? URL에 데이터를 담아 전달하는 방법을 고민하게 되는 순간입니다.

이런 상황은 REST API 개발에서 가장 기본적이면서도 중요한 부분입니다. 경로에 직접 값을 넣을지, 쿼리 스트링으로 보낼지 결정하는 것은 API 설계의 핵심입니다.

잘못 설계하면 API가 직관적이지 않고 사용하기 어려워집니다. 바로 이럴 때 필요한 것이 경로 매개변수와 쿼리 매개변수입니다.

FastAPI는 타입 힌트만으로 이 둘을 명확하게 구분하고 자동 검증까지 제공합니다.

개요

간단히 말해서, 경로 매개변수는 URL 경로의 일부로 전달되는 필수 값이고, 쿼리 매개변수는 ?key=value 형태로 전달되는 선택적 값입니다. 경로 매개변수가 필요한 이유는 RESTful 설계에서 리소스를 식별하기 위함입니다.

/users/123처럼 특정 리소스를 지정할 때 사용합니다. 쿼리 매개변수는 필터링, 정렬, 페이지네이션 같은 부가 옵션을 전달할 때 유용합니다.

예를 들어, /users?page=2&limit=10 같은 경우에 검색 조건을 전달하는 데 사용됩니다. 기존 방식에서는 request.args를 파싱하고 타입 변환과 검증을 수동으로 처리해야 했다면, FastAPI에서는 함수 파라미터로 선언만 하면 모든 처리가 자동화됩니다.

핵심 특징은 첫째, 경로에 중괄호로 감싼 변수는 자동으로 경로 매개변수가 됩니다. 둘째, 함수 파라미터에 기본값을 지정하면 자동으로 쿼리 매개변수가 됩니다.

셋째, 타입 힌트로 자동 변환과 검증이 이루어집니다. 이러한 특징들이 API 엔드포인트를 매우 직관적으로 만들어줍니다.

코드 예제

from fastapi import FastAPI, Query, Path
from typing import Optional

app = FastAPI()

# 경로 매개변수 사용
@app.get("/users/{user_id}")
async def get_user(
    user_id: int = Path(..., ge=1, description="사용자 ID")
):
    # user_id는 자동으로 int로 변환되고 1 이상인지 검증됨
    return {"user_id": user_id, "name": f"User {user_id}"}

# 쿼리 매개변수 사용
@app.get("/items/")
async def list_items(
    skip: int = Query(0, ge=0),  # 기본값 0, 0 이상
    limit: int = Query(10, le=100),  # 기본값 10, 100 이하
    keyword: Optional[str] = None  # 선택적 파라미터
):
    return {"skip": skip, "limit": limit, "keyword": keyword}

설명

이것이 하는 일: URL에서 데이터를 추출하고, 타입 변환 및 검증을 자동으로 수행하여 안전하게 함수에 전달합니다. 첫 번째로, {user_id} 경로 매개변수는 URL 경로의 일부로 받아옵니다.

Path 함수를 사용하면 추가 검증 규칙을 설정할 수 있는데, ge=1은 greater than or equal(1 이상)을 의미합니다. /users/0이나 /users/abc 같은 잘못된 요청은 자동으로 422 에러를 반환합니다.

그 다음으로, list_items 함수의 쿼리 매개변수들은 Query 함수를 통해 더 세밀한 제어가 가능합니다. skip과 limit은 기본값이 있어서 선택적이지만, 타입과 범위 검증은 자동으로 이루어집니다.

keyword는 Optional[str]로 선언되어 있어 None이 허용됩니다. 세 번째로, 실제 요청 시 /items/?skip=20&limit=50&keyword=fastapi처럼 호출하면, FastAPI가 자동으로 문자열을 정수로 변환하고 검증 규칙을 확인합니다.

검증에 실패하면 어떤 필드가 왜 잘못되었는지 상세한 에러 메시지를 JSON으로 반환합니다. 여러분이 이 코드를 사용하면 수동으로 파라미터를 파싱하고 검증하는 보일러플레이트 코드를 완전히 제거할 수 있습니다.

실무에서는 잘못된 입력으로 인한 버그가 크게 줄어들고, API 문서에도 각 파라미터의 타입과 제약사항이 자동으로 표시되어 협업이 수월해집니다.

실전 팁

💡 경로 매개변수는 항상 필수값이므로 리소스 식별자(ID)에만 사용하고, 선택적 값은 쿼리 매개변수로 처리하세요

💡 Query의 min_length, max_length로 문자열 길이를 제한하고, regex로 패턴 검증도 가능합니다

💡 Path의 gt(greater than), lt(less than) 등을 활용하여 숫자 범위를 명확히 제한하면 비즈니스 로직이 간결해집니다

💡 Optional[str] = None 대신 str | None = None (Python 3.10+) 문법을 사용하면 더 현대적인 코드가 됩니다

💡 쿼리 매개변수가 많아지면 Pydantic 모델로 그룹화하여 관리하면 코드 가독성이 향상됩니다

3. Pydantic 모델과 데이터 검증

시작하며

여러분이 회원가입 API를 만들 때, 이메일 형식이 맞는지, 비밀번호가 충분히 강력한지, 나이가 유효한 범위인지 일일이 검증하는 코드를 작성하느라 시간을 낭비한 경험이 있으신가요? 이런 데이터 검증 코드는 비즈니스 로직보다 더 많은 줄을 차지하는 경우가 많습니다.

if문의 연속으로 가독성이 떨어지고, 한 곳에서 수정하면 다른 곳도 일일이 찾아 고쳐야 하는 유지보수 악몽이 펼쳐집니다. 바로 이럴 때 필요한 것이 Pydantic 모델입니다.

타입 힌트와 필드 검증자를 선언만 하면, 모든 검증이 자동으로 처리되고 명확한 에러 메시지까지 제공됩니다.

개요

간단히 말해서, Pydantic은 Python 타입 어노테이션을 사용하여 데이터를 검증하고 파싱하는 라이브러리이며, FastAPI의 핵심 의존성입니다. Pydantic이 필요한 이유는 타입 안전성과 데이터 무결성을 보장하기 때문입니다.

클라이언트로부터 받은 JSON 데이터를 Python 객체로 변환하면서 동시에 모든 필드를 검증합니다. 예를 들어, 사용자 등록 API에서 이메일, 비밀번호, 나이 등을 한 번에 검증하고 변환하는 작업을 한 줄의 타입 힌트로 처리할 수 있습니다.

기존에는 수동으로 각 필드를 체크하고 에러 메시지를 작성했다면, Pydantic은 모델 클래스 정의만으로 모든 검증 로직이 자동 생성됩니다. Pydantic의 핵심 특징은 첫째, Field를 통한 세밀한 검증 규칙 설정이 가능합니다.

둘째, validator 데코레이터로 커스텀 검증 로직을 추가할 수 있습니다. 셋째, 중첩된 모델을 지원하여 복잡한 데이터 구조도 쉽게 처리합니다.

이러한 특징들이 데이터 계층을 견고하게 만들어 런타임 에러를 사전에 방지합니다.

코드 예제

from pydantic import BaseModel, Field, EmailStr, validator
from typing import Optional
from datetime import datetime

class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: EmailStr  # 자동으로 이메일 형식 검증
    password: str = Field(..., min_length=8)
    age: int = Field(..., ge=18, le=120)
    phone: Optional[str] = None

    # 커스텀 검증 로직
    @validator('password')
    def validate_password(cls, v):
        if not any(char.isdigit() for char in v):
            raise ValueError('비밀번호는 최소 1개의 숫자를 포함해야 합니다')
        if not any(char.isupper() for char in v):
            raise ValueError('비밀번호는 최소 1개의 대문자를 포함해야 합니다')
        return v

    class Config:
        schema_extra = {
            "example": {
                "username": "johndoe",
                "email": "john@example.com",
                "password": "SecurePass123",
                "age": 25
            }
        }

@app.post("/register/")
async def register_user(user: UserCreate):
    # 이 시점에서 user는 이미 완전히 검증된 상태
    return {"message": "회원가입 성공", "username": user.username}

설명

이것이 하는 일: 들어오는 요청 데이터를 Pydantic 모델에 정의된 규칙에 따라 자동으로 검증하고, 타입을 변환하며, 커스텀 검증 로직까지 실행합니다. 첫 번째로, UserCreate 클래스는 BaseModel을 상속받아 각 필드의 타입과 제약사항을 선언합니다.

Field 함수의 첫 번째 인자 ...는 필수 필드를 의미하고, min_length, max_length, ge(이상), le(이하) 같은 파라미터로 세밀한 검증 규칙을 설정합니다. EmailStr 타입은 pydantic[email] 패키지를 설치하면 사용할 수 있으며, 이메일 형식을 자동으로 검증합니다.

그 다음으로, @validator 데코레이터를 사용하여 필드별 커스텀 검증 로직을 추가합니다. validate_password 메서드는 비밀번호가 숫자와 대문자를 포함하는지 확인하고, 조건을 만족하지 않으면 ValueError를 발생시킵니다.

이 에러는 FastAPI가 자동으로 422 응답으로 변환하여 클라이언트에게 전달합니다. 세 번째로, Config 클래스의 schema_extra에 예시 데이터를 추가하면, API 문서(/docs)에서 사용자가 테스트할 때 미리 채워진 예시를 볼 수 있습니다.

이것은 API 사용성을 크게 향상시킵니다. 마지막으로, register_user 함수에서 user: UserCreate로 선언하면, FastAPI가 요청 본문을 파싱하여 UserCreate 모델의 모든 검증을 거친 후 함수에 전달합니다.

여러분이 이 코드를 사용하면 함수 내부에서는 이미 완벽하게 검증된 깨끗한 데이터만 다루게 되어, 방어적 프로그래밍 코드가 필요 없어집니다. 실무에서는 보안 취약점이 줄어들고, 디버깅 시간이 대폭 감소합니다.

실전 팁

💡 EmailStr 사용을 위해 pip install pydantic[email]을 설치하세요. 이메일 검증을 정규식으로 직접 구현하는 것보다 안전합니다

💡 민감한 필드(비밀번호 등)는 응답 모델에서 제외하려면 별도의 UserResponse 모델을 만들어 사용하세요

💡 validator의 pre=True 옵션을 사용하면 타입 변환 전에 검증을 수행할 수 있어, 원본 문자열을 정제할 때 유용합니다

💡 여러 필드를 동시에 검증하려면 @root_validator를 사용하세요. 예: 비밀번호와 비밀번호 확인이 일치하는지 검사

💡 orm_mode = True를 Config에 추가하면 SQLAlchemy 모델을 Pydantic 모델로 쉽게 변환할 수 있어 DB 연동이 편리해집니다

4. 비동기 처리와 async/await

시작하며

여러분이 API에서 외부 서비스를 호출하거나 데이터베이스 쿼리를 실행할 때, 하나의 요청이 끝날 때까지 다른 요청들이 대기하는 상황을 본 적 있나요? 동시 접속자가 많아지면 응답 시간이 기하급수적으로 늘어나는 문제가 발생합니다.

이런 병목 현상은 동기 방식의 블로킹 I/O 때문에 발생합니다. 데이터베이스 응답을 기다리는 동안 CPU는 놀고 있지만, 다른 요청을 처리하지 못하고 대기하게 됩니다.

서버 자원은 충분한데 처리량이 낮아지는 비효율적인 상황입니다. 바로 이럴 때 필요한 것이 비동기 처리입니다.

async/await를 사용하면 I/O 작업이 완료되기를 기다리는 동안 다른 요청을 처리하여, 같은 서버 자원으로 훨씬 더 많은 요청을 처리할 수 있습니다.

개요

간단히 말해서, 비동기 처리는 I/O 작업이 완료되기를 기다리는 동안 다른 작업을 수행할 수 있게 하여, 서버의 처리량을 극적으로 향상시키는 프로그래밍 패러다임입니다. 비동기 처리가 필요한 이유는 웹 API의 대부분 시간이 네트워크나 디스크 I/O를 기다리는 데 소비되기 때문입니다.

데이터베이스 쿼리, 외부 API 호출, 파일 읽기/쓰기 같은 작업은 CPU를 거의 사용하지 않고 응답만 기다립니다. 예를 들어, 100ms가 걸리는 DB 쿼리를 동기 방식으로 처리하면 초당 10개 요청밖에 처리 못하지만, 비동기 방식으로 처리하면 수천 개의 요청을 동시에 처리할 수 있습니다.

기존 동기 방식에서는 멀티스레딩이나 멀티프로세싱으로 동시성을 확보했다면, 비동기 방식은 단일 스레드에서 이벤트 루프를 통해 더 효율적으로 동시성을 처리합니다. 비동기 처리의 핵심 특징은 첫째, async def로 정의된 함수는 코루틴이 되어 중단/재개가 가능합니다.

둘째, await 키워드로 I/O 작업을 기다리는 동안 제어권을 이벤트 루프에 반환합니다. 셋째, 컨텍스트 스위칭 비용이 스레드보다 훨씬 적어 고성능을 냅니다.

이러한 특징들이 실시간 채팅, IoT 데이터 수집, 마이크로서비스 간 통신 같은 I/O 집약적 작업에서 빛을 발합니다.

코드 예제

import httpx
from fastapi import FastAPI
import asyncio

app = FastAPI()

# 비동기 데이터베이스 쿼리 시뮬레이션
async def fetch_user_from_db(user_id: int):
    # 실제로는 asyncpg, motor 등 비동기 DB 드라이버 사용
    await asyncio.sleep(0.1)  # DB 쿼리 대기 시뮬레이션
    return {"id": user_id, "name": f"User {user_id}"}

# 비동기 외부 API 호출
async def fetch_external_data(user_id: int):
    async with httpx.AsyncClient() as client:
        # await로 응답을 기다리는 동안 다른 요청 처리 가능
        response = await client.get(f"https://api.example.com/data/{user_id}")
        return response.json()

# 여러 비동기 작업을 병렬로 실행
@app.get("/users/{user_id}/complete")
async def get_user_complete_data(user_id: int):
    # asyncio.gather로 여러 작업을 동시에 실행
    user, external_data = await asyncio.gather(
        fetch_user_from_db(user_id),
        fetch_external_data(user_id)
    )
    return {"user": user, "external_data": external_data}

설명

이것이 하는 일: I/O 작업이 완료되기를 기다리는 동안 이벤트 루프에 제어권을 반환하여, 단일 스레드에서 수천 개의 동시 요청을 효율적으로 처리합니다. 첫 번째로, async def로 정의된 함수들은 코루틴이 되어, await를 만나면 실행이 중단되고 다른 코루틴이 실행될 수 있습니다.

fetch_user_from_db 함수에서 await asyncio.sleep(0.1)를 실행하면, 0.1초 동안 CPU를 놀리지 않고 다른 요청의 코루틴으로 전환됩니다. 실제 프로덕션에서는 asyncpg(PostgreSQL), motor(MongoDB) 같은 비동기 데이터베이스 드라이버를 사용합니다.

그 다음으로, httpx.AsyncClient는 비동기 HTTP 요청을 위한 클라이언트입니다. requests 라이브러리는 동기 방식이라 FastAPI의 비동기 처리와 맞지 않으므로, httpx나 aiohttp를 사용해야 합니다.

await client.get()을 호출하면 네트워크 응답을 기다리는 동안 다른 요청을 처리할 수 있어, 서버 처리량이 10배 이상 향상될 수 있습니다. 세 번째로, asyncio.gather는 여러 코루틴을 동시에 실행하고 모든 결과가 완료될 때까지 기다립니다.

위 예제에서 DB 쿼리와 외부 API 호출을 순차적으로 실행하면 0.1초 + 네트워크 시간이 걸리지만, gather를 사용하면 둘 중 느린 작업의 시간만 걸립니다. 실제로 여러 마이크로서비스를 호출해야 하는 경우 응답 시간을 절반 이하로 줄일 수 있습니다.

여러분이 이 코드를 사용하면 동일한 하드웨어에서 처리할 수 있는 동시 요청 수가 극적으로 증가합니다. 실무에서는 클라우드 비용을 크게 절감할 수 있고, 사용자는 더 빠른 응답을 경험하게 됩니다.

특히 MSA 환경에서 여러 서비스를 조합해야 할 때 비동기 처리는 필수입니다.

실전 팁

💡 동기 라이브러리(requests, psycopg2)를 async 함수에서 사용하면 오히려 성능이 떨어집니다. 반드시 비동기 라이브러리(httpx, asyncpg)를 사용하세요

💡 CPU 집약적 작업(이미지 처리, 암호화)은 비동기로 이점이 없습니다. 이런 경우 def를 사용하거나 ProcessPoolExecutor를 고려하세요

💡 asyncio.create_task()로 백그라운드 작업을 시작하고 응답을 먼저 반환할 수 있어, 사용자 경험을 개선할 수 있습니다

💡 비동기 컨텍스트 매니저(async with)를 사용하여 리소스를 안전하게 관리하세요. 특히 DB 커넥션과 HTTP 클라이언트에서 중요합니다

💡 디버깅 시 asyncio.run()의 debug=True 옵션을 활성화하면 블로킹 I/O나 느린 코루틴을 감지할 수 있습니다

5. 의존성 주입 시스템

시작하며

여러분이 모든 API 엔드포인트에서 데이터베이스 연결을 만들거나, 사용자 인증을 체크하거나, 로깅을 설정하는 코드를 반복해서 작성하고 있나요? 코드 중복이 심해지고 공통 로직을 수정하려면 수십 개 파일을 고쳐야 하는 상황이 발생합니다.

이런 문제는 횡단 관심사(cross-cutting concerns)를 제대로 처리하지 못해서 생깁니다. 인증, 권한 확인, DB 세션 관리, 로깅 같은 공통 기능이 비즈니스 로직과 뒤섞여 코드가 복잡해지고 테스트하기도 어려워집니다.

바로 이럴 때 필요한 것이 의존성 주입 시스템입니다. FastAPI의 Depends를 사용하면 공통 로직을 재사용 가능한 컴포넌트로 분리하고, 자동으로 실행되게 할 수 있습니다.

개요

간단히 말해서, 의존성 주입은 함수나 클래스가 필요로 하는 의존성을 외부에서 주입받아 사용하는 디자인 패턴이며, FastAPI는 이를 Depends를 통해 우아하게 구현합니다. 의존성 주입이 필요한 이유는 코드 재사용성과 테스트 용이성을 높이기 때문입니다.

인증 로직, DB 세션, 현재 사용자 정보 같은 공통 요소를 한 곳에서 관리하고 필요한 곳에서 선언만 하면 자동으로 주입됩니다. 예를 들어, 로그인한 사용자만 접근할 수 있는 엔드포인트를 만들 때, 매번 토큰을 파싱하고 검증하는 코드를 작성하지 않고 Depends(get_current_user)만 추가하면 됩니다.

기존 방식에서는 전역 변수나 데코레이터로 공통 로직을 처리했다면, FastAPI의 의존성 주입은 타입 시스템과 통합되어 IDE 자동완성과 타입 체크의 이점을 모두 누릴 수 있습니다. 의존성 주입의 핵심 특징은 첫째, 의존성도 다른 의존성을 가질 수 있어 계층 구조를 만들 수 있습니다.

둘째, 캐싱을 통해 같은 요청 내에서 의존성을 재사용할 수 있습니다. 셋째, 의존성은 자동으로 OpenAPI 문서에 반영됩니다.

이러한 특징들이 대규모 애플리케이션의 아키텍처를 깔끔하게 유지시켜 줍니다.

코드 예제

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from typing import AsyncGenerator
import jwt

security = HTTPBearer()

# DB 세션 의존성
async def get_db() -> AsyncGenerator:
    # 실제로는 SQLAlchemy AsyncSession 사용
    db = "DB Connection"
    try:
        yield db  # yield로 의존성 제공
    finally:
        # 요청이 끝나면 자동으로 정리
        print("DB 세션 종료")

# 인증 의존성
async def get_current_user(
    credentials: HTTPAuthorizationCredentials = Depends(security),
    db = Depends(get_db)  # 의존성이 다른 의존성을 의존
):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, "secret_key", algorithms=["HS256"])
        user_id = payload.get("user_id")
        if user_id is None:
            raise HTTPException(status_code=401, detail="유효하지 않은 토큰")
        # db를 사용하여 사용자 정보 조회
        return {"id": user_id, "name": "John"}
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="토큰 검증 실패")

# 엔드포인트에서 의존성 사용
@app.get("/protected/")
async def protected_route(
    current_user: dict = Depends(get_current_user)
):
    # current_user는 자동으로 주입됨
    return {"message": f"안녕하세요, {current_user['name']}님"}

설명

이것이 하는 일: 엔드포인트 함수가 실행되기 전에 필요한 의존성들을 자동으로 실행하고, 그 결과를 함수 파라미터로 주입하여 코드 중복을 제거하고 관심사를 분리합니다. 첫 번째로, get_db 함수는 yield를 사용하는 제너레이터 의존성입니다.

yield 이전 코드는 요청 처리 전에 실행되고(DB 연결 생성), yield 이후 코드는 요청 처리 후에 실행되어(DB 연결 정리) 리소스 관리를 안전하게 할 수 있습니다. 이것은 컨텍스트 매니저와 비슷한 역할을 하며, 예외가 발생해도 finally 블록이 실행되어 리소스 누수를 방지합니다.

그 다음으로, get_current_user 의존성은 HTTPBearer 보안 스킴을 통해 Authorization 헤더에서 토큰을 자동으로 추출합니다. 이 의존성은 또 다른 의존성인 get_db를 가지고 있어, FastAPI가 자동으로 실행 순서를 결정하고 get_db를 먼저 실행한 후 그 결과를 get_current_user에 전달합니다.

JWT 토큰을 검증하고 사용자 정보를 조회하는 복잡한 로직이 하나의 의존성으로 캡슐화됩니다. 세 번째로, protected_route 엔드포인트는 Depends(get_current_user)만 선언하면 자동으로 인증이 처리됩니다.

FastAPI는 요청이 들어오면 의존성 트리를 순회하며 security -> get_db -> get_current_user 순으로 실행하고, 최종 결과를 current_user 파라미터에 주입합니다. 인증에 실패하면 HTTPException이 발생하여 엔드포인트 함수는 실행되지 않습니다.

여러분이 이 코드를 사용하면 인증이 필요한 모든 엔드포인트에서 단 한 줄(Depends(get_current_user))만 추가하면 됩니다. 실무에서는 인증 로직이 변경되어도 get_current_user 함수만 수정하면 모든 엔드포인트에 자동 반영되고, 테스트 시에는 의존성을 mock으로 교체하여 쉽게 유닛 테스트를 작성할 수 있습니다.

실전 팁

💡 의존성은 app.dependency_overrides로 테스트 시 mock으로 교체할 수 있어, 통합 테스트가 매우 쉬워집니다

💡 같은 요청 내에서 의존성은 캐시되므로, Depends(get_db)를 여러 곳에서 사용해도 한 번만 실행됩니다. use_cache=False로 비활성화 가능합니다

💡 라우터 레벨에서 dependencies 파라미터로 공통 의존성을 설정하면 모든 엔드포인트에 자동 적용됩니다

💡 클래스 기반 의존성을 만들면 상태를 유지하면서 재사용할 수 있어, 페이지네이션이나 필터링 로직에 유용합니다

💡 BackgroundTasks 의존성을 사용하면 응답을 반환한 후 백그라운드에서 이메일 전송 같은 작업을 수행할 수 있습니다

6. 에러 핸들링과 HTTP 예외

시작하며

여러분이 API를 개발할 때, 리소스를 찾을 수 없거나, 권한이 없거나, 잘못된 요청이 들어왔을 때 어떻게 에러를 반환하시나요? 에러 메시지가 일관성 없고, 클라이언트가 무슨 일이 일어났는지 이해하기 어려운 경우가 많습니다.

이런 문제는 에러 처리 전략이 없어서 발생합니다. 어떤 곳에서는 문자열로 에러를 반환하고, 어떤 곳에서는 JSON으로 반환하며, 상태 코드도 제각각입니다.

프론트엔드 개발자는 각 에러 케이스를 일일이 확인하며 처리 로직을 작성해야 합니다. 바로 이럴 때 필요한 것이 체계적인 에러 핸들링입니다.

FastAPI의 HTTPException과 커스텀 예외 핸들러를 사용하면 일관된 형식으로 명확한 에러 메시지를 제공할 수 있습니다.

개요

간단히 말해서, 에러 핸들링은 예외 상황을 적절한 HTTP 상태 코드와 명확한 메시지로 변환하여 클라이언트가 문제를 이해하고 대응할 수 있게 하는 것입니다. 에러 핸들링이 필요한 이유는 API의 사용성과 디버깅 효율성을 높이기 때문입니다.

명확한 에러 메시지는 클라이언트 개발자가 문제를 빠르게 파악하고 해결하도록 돕습니다. 예를 들어, "User not found with id: 123"이라는 메시지는 "Internal Server Error"보다 훨씬 유용합니다.

또한 적절한 HTTP 상태 코드(404, 401, 403 등)를 사용하면 클라이언트가 조건부로 에러를 처리할 수 있습니다. 기존 방식에서는 try-except 블록에서 수동으로 에러 응답을 만들었다면, FastAPI에서는 HTTPException을 발생시키기만 하면 자동으로 적절한 형식의 JSON 응답으로 변환됩니다.

에러 핸들링의 핵심 특징은 첫째, HTTPException으로 표준 HTTP 에러를 쉽게 반환할 수 있습니다. 둘째, 커스텀 예외 핸들러로 애플리케이션 전체의 에러 형식을 통일할 수 있습니다.

셋째, status 모듈로 상태 코드를 의미 있는 상수로 사용할 수 있습니다. 이러한 특징들이 API를 더 전문적이고 사용하기 쉽게 만듭니다.

코드 예제

from fastapi import FastAPI, HTTPException, status, Request
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# 커스텀 예외 정의
class ItemNotFoundException(Exception):
    def __init__(self, item_id: int):
        self.item_id = item_id

# 커스텀 예외 핸들러 등록
@app.exception_handler(ItemNotFoundException)
async def item_not_found_handler(request: Request, exc: ItemNotFoundException):
    return JSONResponse(
        status_code=status.HTTP_404_NOT_FOUND,
        content={
            "error": "Item Not Found",
            "message": f"아이템 ID {exc.item_id}를 찾을 수 없습니다",
            "item_id": exc.item_id
        }
    )

# Pydantic 검증 에러 커스터마이징
@app.exception_handler(RequestValidationError)
async def validation_exception_handler(request: Request, exc: RequestValidationError):
    return JSONResponse(
        status_code=status.HTTP_422_UNPROCESSABLE_ENTITY,
        content={
            "error": "Validation Error",
            "message": "요청 데이터가 유효하지 않습니다",
            "details": exc.errors()
        }
    )

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    # 표준 HTTP 예외 사용
    if item_id < 1:
        raise HTTPException(
            status_code=status.HTTP_400_BAD_REQUEST,
            detail="item_id는 1 이상이어야 합니다"
        )

    # 커스텀 예외 사용
    if item_id == 999:
        raise ItemNotFoundException(item_id=item_id)

    return {"item_id": item_id, "name": "Sample Item"}

설명

이것이 하는 일: 예외가 발생하면 자동으로 적절한 HTTP 응답으로 변환하고, 커스텀 예외 핸들러를 통해 일관된 에러 형식을 제공하여 API의 사용성을 높입니다. 첫 번째로, ItemNotFoundException이라는 도메인 특화 예외를 정의합니다.

이것은 Python의 일반 Exception을 상속받으며, 비즈니스 로직에서 발생하는 특정 상황을 표현합니다. item_id를 속성으로 저장하여 에러 메시지에 컨텍스트를 제공할 수 있습니다.

이렇게 도메인 예외를 정의하면 코드에서 "무엇이 잘못되었는지"를 명확히 표현할 수 있습니다. 그 다음으로, @app.exception_handler 데코레이터로 ItemNotFoundException이 발생했을 때 실행될 핸들러를 등록합니다.

이 핸들러는 예외를 받아 JSONResponse로 변환하는데, status_code와 content를 명시적으로 설정하여 일관된 형식의 에러 응답을 만듭니다. 이렇게 하면 애플리케이션 전체에서 동일한 예외가 항상 같은 형식으로 반환됩니다.

세 번째로, RequestValidationError 핸들러는 Pydantic의 검증 에러를 커스터마이징합니다. 기본 검증 에러 메시지는 기술적이고 이해하기 어려울 수 있는데, 이를 사용자 친화적인 메시지로 변환할 수 있습니다.

exc.errors()는 어떤 필드가 왜 잘못되었는지 상세 정보를 포함하므로, 프론트엔드에서 필드별로 에러를 표시할 수 있습니다. 마지막으로, 엔드포인트에서는 HTTPException으로 간단한 에러를, ItemNotFoundException으로 도메인 특화 에러를 발생시킵니다.

status 모듈의 상수를 사용하면 400, 404 같은 숫자 대신 HTTP_400_BAD_REQUEST처럼 의미 있는 이름을 사용할 수 있어 코드 가독성이 높아집니다. 여러분이 이 코드를 사용하면 클라이언트 개발자가 에러를 쉽게 이해하고 처리할 수 있으며, 실무에서는 Sentry 같은 모니터링 도구와 통합하여 에러를 추적하고 분석할 수 있습니다.

실전 팁

💡 status 모듈의 상수(HTTP_404_NOT_FOUND 등)를 사용하면 매직 넘버를 피하고 코드 의도를 명확히 할 수 있습니다

💡 에러 응답에 request_id를 포함하면 로그와 연결하여 디버깅이 훨씬 쉬워집니다. 미들웨어에서 자동으로 추가할 수 있습니다

💡 프로덕션 환경에서는 내부 에러(500)의 상세 정보를 숨기고 로그에만 남겨 보안을 강화하세요

💡 HTTPException의 headers 파라미터로 WWW-Authenticate 같은 커스텀 헤더를 추가할 수 있습니다

💡 비즈니스 로직별로 예외 클래스를 계층화(BaseBusinessException -> UserNotFoundException 등)하면 에러 처리가 체계적으로 관리됩니다

7. 미들웨어와 요청/응답 처리

시작하며

여러분이 모든 API 요청에 대해 로깅을 하거나, CORS 설정을 추가하거나, 요청 시간을 측정하고 싶은데 각 엔드포인트마다 코드를 추가하는 것은 비효율적이라고 느낀 적 있나요? 이런 횡단 관심사는 애플리케이션 전체에 영향을 미치지만, 비즈니스 로직과는 분리되어야 합니다.

각 엔드포인트에 중복 코드를 추가하면 유지보수가 어려워지고, 일관성을 유지하기도 힘듭니다. 바로 이럴 때 필요한 것이 미들웨어입니다.

미들웨어는 모든 요청과 응답을 가로채서 공통 처리를 수행할 수 있게 해주며, 애플리케이션 아키텍처를 깔끔하게 유지시켜 줍니다.

개요

간단히 말해서, 미들웨어는 요청이 엔드포인트에 도달하기 전과 응답이 클라이언트에 전달되기 전에 실행되는 레이어로, 횡단 관심사를 처리하는 강력한 메커니즘입니다. 미들웨어가 필요한 이유는 인증, 로깅, CORS, 압축, 요청 시간 측정 같은 공통 기능을 한 곳에서 관리할 수 있기 때문입니다.

각 엔드포인트는 비즈니스 로직에만 집중하고, 미들웨어가 인프라 관심사를 처리합니다. 예를 들어, 모든 API 응답 시간을 헤더에 추가하거나, 요청/응답을 로깅하거나, 특정 IP를 차단하는 작업을 미들웨어에서 일괄 처리할 수 있습니다.

기존 Flask에서는 before_request, after_request 데코레이터를 사용했다면, FastAPI에서는 ASGI 미들웨어 표준을 따라 더 강력하고 유연한 처리가 가능합니다. 미들웨어의 핵심 특징은 첫째, 요청 전후에 코드를 실행할 수 있는 양방향 처리가 가능합니다.

둘째, 여러 미들웨어를 체인으로 연결하여 순차적으로 실행됩니다. 셋째, 요청을 조기에 차단하거나 응답을 수정할 수 있습니다.

이러한 특징들이 애플리케이션 전체의 동작을 제어하는 강력한 도구가 됩니다.

코드 예제

from fastapi import FastAPI, Request
from fastapi.middleware.cors import CORSMiddleware
import time
import logging

app = FastAPI()

# CORS 미들웨어 추가
app.add_middleware(
    CORSMiddleware,
    allow_origins=["http://localhost:3000"],  # 프론트엔드 도메인
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 커스텀 미들웨어 - 요청 시간 측정
@app.middleware("http")
async def add_process_time_header(request: Request, call_next):
    # 요청 처리 전
    start_time = time.time()
    request_id = f"{start_time}-{request.client.host}"

    logging.info(f"[{request_id}] {request.method} {request.url}")

    # 다음 미들웨어 또는 엔드포인트 실행
    response = await call_next(request)

    # 응답 처리 후
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    response.headers["X-Request-ID"] = request_id

    logging.info(f"[{request_id}] Completed in {process_time:.3f}s")

    return response

@app.get("/")
async def root():
    return {"message": "Hello World"}

설명

이것이 하는 일: 요청이 엔드포인트에 도달하기 전과 응답이 클라이언트에 반환되기 전에 코드를 실행하여, 애플리케이션 전체에 영향을 미치는 공통 기능을 처리합니다. 첫 번째로, CORSMiddleware는 FastAPI에 내장된 미들웨어로, Cross-Origin Resource Sharing을 처리합니다.

allow_origins에 허용할 도메인을 지정하면, 프론트엔드 애플리케이션이 다른 도메인에서 실행되어도 API를 호출할 수 있습니다. allow_credentials=True는 쿠키를 포함한 요청을 허용하며, allow_methods와 allow_headers는 허용할 HTTP 메서드와 헤더를 지정합니다.

프로덕션에서는 "*" 대신 구체적인 값을 사용해야 보안이 강화됩니다. 그 다음으로, @app.middleware("http") 데코레이터로 커스텀 미들웨어를 정의합니다.

이 함수는 request와 call_next 두 개의 파라미터를 받는데, call_next는 다음 미들웨어나 엔드포인트를 실행하는 함수입니다. call_next를 호출하기 전의 코드는 요청 처리 전에 실행되고, call_next를 호출한 후의 코드는 응답 처리 후에 실행됩니다.

세 번째로, 요청 시간을 측정하기 위해 start_time을 기록하고, call_next를 await하여 실제 엔드포인트를 실행합니다. 엔드포인트가 완료되면 response 객체를 받아 처리 시간을 계산하고, X-Process-Time 커스텀 헤더를 추가합니다.

이렇게 하면 클라이언트나 모니터링 도구에서 각 요청의 응답 시간을 추적할 수 있습니다. 마지막으로, request_id를 생성하여 로깅에 사용하고 응답 헤더에도 포함시킵니다.

이것은 분산 추적(distributed tracing)의 기본이 되며, 프론트엔드에서 에러가 발생했을 때 request_id를 제공하면 백엔드 로그에서 빠르게 찾을 수 있습니다. 여러분이 이 코드를 사용하면 모든 엔드포인트에 일관된 로깅과 모니터링이 자동으로 적용되며, 실무에서는 성능 병목을 찾거나 사용자 행동을 분석하는 데 매우 유용합니다.

실전 팁

💡 미들웨어는 등록 순서의 역순으로 응답을 처리합니다. 먼저 등록한 미들웨어가 가장 바깥쪽 레이어가 됩니다

💡 GZipMiddleware를 추가하면 큰 JSON 응답을 자동으로 압축하여 네트워크 사용량을 줄일 수 있습니다

💡 특정 경로를 미들웨어에서 제외하려면 request.url.path를 체크하여 조건부로 처리하세요. 예: /health, /metrics

💡 미들웨어에서 예외가 발생하면 클라이언트에 500 에러가 반환됩니다. try-except로 감싸서 안전하게 처리하세요

💡 Starlette의 BaseHTTPMiddleware 대신 순수 ASGI 미들웨어를 사용하면 성능이 더 좋지만 구현이 복잡합니다. 성능이 중요한 경우 고려하세요

8. 데이터베이스 연동 (SQLAlchemy)

시작하며

여러분이 API를 만들 때 데이터를 어딘가에 영구적으로 저장해야 하는데, SQL 쿼리를 문자열로 작성하다가 오타나 SQL 인젝션 취약점을 만든 경험이 있으신가요? 이런 문제는 데이터베이스를 직접 다루면서 발생합니다.

테이블 구조가 변경되면 모든 SQL 쿼리를 찾아 수정해야 하고, 타입 안전성도 보장되지 않아 런타임 에러가 자주 발생합니다. 코드와 데이터베이스 스키마가 따로 관리되어 싱크가 맞지 않는 경우도 많습니다.

바로 이럴 때 필요한 것이 ORM(Object-Relational Mapping)입니다. SQLAlchemy를 사용하면 Python 클래스로 테이블을 정의하고, 타입 안전한 쿼리를 작성하며, 데이터베이스 마이그레이션까지 관리할 수 있습니다.

개요

간단히 말해서, SQLAlchemy는 Python의 가장 강력한 ORM 라이브러리로, 데이터베이스 테이블을 Python 클래스로 매핑하여 객체 지향적으로 데이터를 다룰 수 있게 합니다. SQLAlchemy가 필요한 이유는 생산성과 안전성을 동시에 높이기 때문입니다.

SQL을 직접 작성하지 않고 Python 코드로 쿼리를 표현하면, IDE의 자동완성과 타입 체크를 활용할 수 있습니다. 예를 들어, User 테이블에서 이메일로 검색하는 쿼리를 session.query(User).filter(User.email == "test@example.com")처럼 작성하면, 오타가 있으면 IDE가 즉시 알려줍니다.

기존 방식에서는 cursor.execute("SELECT * FROM users WHERE email = ?", (email,))처럼 문자열 쿼리를 사용했다면, SQLAlchemy는 Python 객체와 메서드로 표현하여 더 안전하고 유지보수하기 쉬운 코드를 만듭니다. SQLAlchemy의 핵심 특징은 첫째, 선언적 베이스를 통해 테이블 스키마를 클래스로 정의합니다.

둘째, 세션을 통해 데이터베이스 트랜잭션을 관리합니다. 셋째, 관계(relationship)를 통해 JOIN 쿼리를 객체 탐색으로 단순화합니다.

이러한 특징들이 복잡한 데이터베이스 작업을 직관적으로 만들어 줍니다.

코드 예제

from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker, Session
from fastapi import Depends

# 데이터베이스 연결 설정
DATABASE_URL = "sqlite:///./test.db"
engine = create_engine(DATABASE_URL, connect_args={"check_same_thread": False})
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

Base = declarative_base()

# 모델 정의
class User(Base):
    __tablename__ = "users"

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

# 테이블 생성
Base.metadata.create_all(bind=engine)

# DB 세션 의존성
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# API 엔드포인트
@app.post("/users/")
def create_user(name: str, email: str, age: int, db: Session = Depends(get_db)):
    # ORM을 사용한 데이터 생성
    db_user = User(name=name, email=email, age=age)
    db.add(db_user)
    db.commit()
    db.refresh(db_user)  # DB에서 생성된 ID를 가져옴
    return db_user

@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    # 타입 안전한 쿼리
    user = db.query(User).filter(User.id == user_id).first()
    if user is None:
        raise HTTPException(status_code=404, detail="User not found")
    return user

설명

이것이 하는 일: 데이터베이스 테이블을 Python 클래스로 정의하고, 객체를 통해 데이터를 조작하면 SQLAlchemy가 자동으로 SQL 쿼리를 생성하고 실행합니다. 첫 번째로, create_engine으로 데이터베이스 연결을 생성하고, SessionLocal로 세션 팩토리를 만듭니다.

세션은 데이터베이스와의 대화를 관리하는 객체로, 트랜잭션의 시작과 종료를 담당합니다. SQLite를 예제로 사용했지만, DATABASE_URL을 변경하면 PostgreSQL, MySQL 등 다른 데이터베이스도 동일한 코드로 사용할 수 있습니다.

그 다음으로, User 클래스는 declarative_base를 상속받아 데이터베이스 테이블을 정의합니다. __tablename__으로 테이블 이름을 지정하고, Column으로 각 필드의 타입과 제약사항을 선언합니다.

primary_key=True는 기본 키를, index=True는 인덱스를, unique=True는 유니크 제약을 의미합니다. Base.metadata.create_all()을 호출하면 정의된 모든 테이블이 데이터베이스에 생성됩니다.

세 번째로, get_db 의존성 함수는 요청마다 새로운 세션을 생성하고, 요청이 끝나면 자동으로 close합니다. yield를 사용하여 리소스를 안전하게 관리하며, 예외가 발생해도 finally 블록에서 세션을 닫아 커넥션 누수를 방지합니다.

마지막으로, create_user 엔드포인트에서는 User 객체를 생성하고 db.add()로 세션에 추가한 후, db.commit()으로 실제 데이터베이스에 저장합니다. db.refresh()는 데이터베이스에서 자동 생성된 ID를 객체에 다시 로드합니다.

get_user에서는 db.query(User).filter()로 타입 안전한 쿼리를 작성하며, User.id == user_id 같은 Python 표현식이 자동으로 SQL WHERE 절로 변환됩니다. 여러분이 이 코드를 사용하면 SQL 인젝션 걱정 없이 안전하게 데이터베이스를 다룰 수 있으며, 실무에서는 Alembic으로 스키마 마이그레이션을 관리하여 데이터베이스 변경을 추적하고 버전 관리할 수 있습니다.

실전 팁

💡 프로덕션에서는 SQLAlchemy 2.0의 AsyncSession을 사용하여 비동기 데이터베이스 작업을 수행하세요. asyncpg, aiomysql 등 비동기 드라이버가 필요합니다

💡 Alembic으로 마이그레이션을 관리하면 스키마 변경을 버전 관리하고 롤백할 수 있어 프로덕션 배포가 안전해집니다

💡 N+1 쿼리 문제를 방지하려면 joinedload, selectinload 같은 eager loading을 사용하세요. 성능이 크게 향상됩니다

💡 트랜잭션이 필요한 복잡한 로직은 db.begin()으로 명시적 트랜잭션을 사용하여 원자성을 보장하세요

💡 커넥션 풀 설정(pool_size, max_overflow)을 조정하여 동시 접속 수에 맞는 최적의 성능을 얻으세요

9. 인증과 JWT 토큰

시작하며

여러분이 로그인 기능을 구현할 때, 사용자가 인증된 상태를 어떻게 유지하고, 매 요청마다 어떻게 사용자를 식별해야 할지 고민하신 적 있나요? 세션을 사용하면 서버 메모리를 많이 차지하고, 멀티 서버 환경에서 복잡해집니다.

이런 문제는 전통적인 세션 기반 인증의 한계입니다. 서버가 세션 정보를 메모리나 Redis에 저장해야 하고, 서버가 재시작되면 모든 사용자가 로그아웃됩니다.

마이크로서비스 아키텍처에서는 각 서비스가 세션을 공유하기 위해 추가 인프라가 필요합니다. 바로 이럴 때 필요한 것이 JWT(JSON Web Token) 기반 인증입니다.

토큰 자체에 사용자 정보가 담겨 있어 서버가 상태를 저장하지 않고도 사용자를 인증할 수 있습니다.

개요

간단히 말해서, JWT는 JSON 형태의 정보를 안전하게 전달하기 위한 토큰 방식으로, 서명을 통해 위변조를 방지하고 서버가 상태를 저장하지 않아도 인증을 처리할 수 있습니다. JWT가 필요한 이유는 확장 가능하고 stateless한 인증 시스템을 구축할 수 있기 때문입니다.

서버는 토큰을 발급한 후 저장할 필요가 없으며, 토큰을 받으면 서명만 검증하여 유효성을 확인합니다. 예를 들어, 로그인 시 JWT를 발급하면 클라이언트는 이를 로컬 스토리지에 저장하고, 이후 모든 API 요청에 포함시켜 인증합니다.

기존 세션 방식에서는 서버가 세션 ID를 키로 사용자 정보를 저장했다면, JWT는 토큰 자체에 사용자 정보를 포함시켜 서버가 상태를 관리하지 않아도 됩니다. JWT의 핵심 특징은 첫째, 헤더(알고리즘), 페이로드(데이터), 서명(검증) 세 부분으로 구성됩니다.

둘째, 비밀 키로 서명되어 위변조가 불가능합니다. 셋째, 만료 시간(exp)을 설정하여 토큰의 유효 기간을 제한할 수 있습니다.

이러한 특징들이 현대적인 웹 애플리케이션과 모바일 앱의 표준 인증 방식이 되었습니다.

코드 예제

from fastapi import Depends, HTTPException, status
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
from passlib.context import CryptContext
from datetime import datetime, timedelta
import jwt

SECRET_KEY = "your-secret-key-keep-it-secret"
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

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

# 비밀번호 해싱
def hash_password(password: str) -> str:
    return pwd_context.hash(password)

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

# JWT 토큰 생성
def create_access_token(data: dict):
    to_encode = data.copy()
    expire = datetime.utcnow() + timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# 토큰에서 사용자 정보 추출
def get_current_user(credentials: HTTPAuthorizationCredentials = Depends(security)):
    token = credentials.credentials
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        user_id: int = payload.get("user_id")
        if user_id is None:
            raise HTTPException(status_code=401, detail="유효하지 않은 토큰")
        return {"user_id": user_id, "email": payload.get("email")}
    except jwt.ExpiredSignatureError:
        raise HTTPException(status_code=401, detail="토큰이 만료되었습니다")
    except jwt.JWTError:
        raise HTTPException(status_code=401, detail="토큰 검증 실패")

# 로그인 엔드포인트
@app.post("/login/")
def login(email: str, password: str):
    # 실제로는 DB에서 사용자 조회
    fake_hashed_password = hash_password("secret123")
    if not verify_password(password, fake_hashed_password):
        raise HTTPException(status_code=401, detail="이메일 또는 비밀번호가 잘못되었습니다")

    access_token = create_access_token(data={"user_id": 1, "email": email})
    return {"access_token": access_token, "token_type": "bearer"}

# 보호된 엔드포인트
@app.get("/me/")
def get_me(current_user: dict = Depends(get_current_user)):
    return current_user

설명

이것이 하는 일: 로그인 시 사용자 정보를 담은 JWT 토큰을 발급하고, 이후 요청에서 토큰을 검증하여 서버 상태 없이 사용자를 인증합니다. 첫 번째로, passlib의 CryptContext를 사용하여 비밀번호를 안전하게 해싱합니다.

bcrypt 알고리즘은 일방향 해시로, 같은 비밀번호도 매번 다른 해시값을 생성하지만 verify_password로 검증할 수 있습니다. 비밀번호를 평문으로 저장하는 것은 치명적인 보안 취약점이므로, 반드시 해싱해야 합니다.

그 다음으로, create_access_token 함수는 사용자 정보를 담은 페이로드에 만료 시간(exp)을 추가하고, SECRET_KEY로 서명하여 JWT를 생성합니다. jwt.encode는 헤더, 페이로드, 서명을 base64로 인코딩한 문자열을 반환하며, 이것이 클라이언트에게 전달됩니다.

만료 시간을 설정하면 토큰이 탈취되어도 일정 시간 후 자동으로 무효화되어 보안이 강화됩니다. 세 번째로, get_current_user 의존성은 Authorization 헤더에서 토큰을 추출하고, jwt.decode로 서명을 검증하며 페이로드를 파싱합니다.

SECRET_KEY가 다르거나 토큰이 위변조되었으면 JWTError가 발생하고, 만료 시간이 지났으면 ExpiredSignatureError가 발생합니다. 각 예외를 적절한 HTTP 401 에러로 변환하여 클라이언트가 재인증하도록 유도합니다.

마지막으로, login 엔드포인트는 이메일과 비밀번호를 받아 검증하고, 성공하면 JWT를 발급하여 반환합니다. 클라이언트는 이 토큰을 저장하고, 이후 모든 요청의 Authorization 헤더에 "Bearer {token}" 형식으로 포함시킵니다.

/me 엔드포인트는 Depends(get_current_user)로 자동으로 인증을 요구하며, 인증된 사용자 정보를 받아 사용합니다. 여러분이 이 코드를 사용하면 세션 서버나 Redis 없이도 확장 가능한 인증 시스템을 구축할 수 있으며, 실무에서는 refresh token을 추가하여 access token을 주기적으로 갱신하고, OAuth2를 통합하여 소셜 로그인도 지원할 수 있습니다.

실전 팁

💡 SECRET_KEY는 환경 변수로 관리하고 절대 코드에 하드코딩하지 마세요. 최소 32자 이상의 무작위 문자열을 사용하세요

💡 access token은 짧게(15-30분), refresh token은 길게(7-30일) 설정하여 보안과 사용성의 균형을 맞추세요

💡 토큰에 민감한 정보(비밀번호 등)를 담지 마세요. JWT는 암호화가 아니라 서명만 되어 있어 누구나 디코딩할 수 있습니다

💡 로그아웃 기능을 구현하려면 토큰 블랙리스트를 Redis에 저장하여 해당 토큰을 거부하도록 처리하세요

💡 프로덕션에서는 HTTPS를 필수로 사용하여 토큰이 네트워크에서 탈취되는 것을 방지하세요

10. 백그라운드 작업과 Celery 연동

시작하며

여러분이 회원가입 후 환영 이메일을 보내거나, 대용량 파일을 처리하거나, 정기적으로 데이터를 동기화해야 할 때, 이런 작업들이 완료될 때까지 사용자가 기다려야 한다면 어떨까요? 응답 시간이 길어지고 사용자 경험이 나빠집니다.

이런 문제는 무거운 작업을 동기적으로 처리하면서 발생합니다. 이메일 발송, 이미지 리사이징, 데이터 분석 같은 작업은 시간이 오래 걸리지만 즉시 결과가 필요하지 않습니다.

이런 작업이 API 응답을 지연시키면 타임아웃이 발생하거나 서버 리소스가 낭비됩니다. 바로 이럴 때 필요한 것이 백그라운드 작업 시스템입니다.

FastAPI의 BackgroundTasks로 간단한 작업을, Celery로 복잡한 분산 작업을 처리하여 사용자는 즉시 응답을 받고 작업은 뒤에서 비동기로 처리됩니다.

개요

간단히 말해서, 백그라운드 작업은 API 응답을 반환한 후 별도의 프로세스나 워커에서 실행되는 작업으로, 사용자 응답 시간을 최소화하면서 무거운 작업을 처리할 수 있게 합니다. 백그라운드 작업이 필요한 이유는 사용자 경험과 서버 효율성을 동시에 개선하기 때문입니다.

회원가입 API는 사용자 정보만 저장하고 즉시 응답하며, 환영 이메일은 백그라운드에서 발송합니다. 예를 들어, 프로필 사진 업로드 시 원본은 즉시 저장하고 응답하며, 여러 크기로 리사이징하는 작업은 백그라운드에서 처리할 수 있습니다.

기존 방식에서는 모든 작업을 순차적으로 처리하여 응답 시간이 길었다면, 백그라운드 작업은 필수 작업만 완료하고 응답하여 API 성능을 극적으로 향상시킵니다. 백그라운드 작업의 핵심 특징은 첫째, FastAPI의 BackgroundTasks는 간단한 작업을 응답 후 실행합니다.

둘째, Celery는 메시지 큐(Redis, RabbitMQ)를 통해 분산 워커에서 작업을 처리합니다. 셋째, 재시도, 스케줄링, 모니터링 같은 고급 기능을 제공합니다.

이러한 특징들이 복잡한 비즈니스 로직을 견고하게 처리할 수 있게 해줍니다.

코드 예제

from fastapi import BackgroundTasks, FastAPI
from celery import Celery
import time

app = FastAPI()

# Celery 설정
celery_app = Celery(
    "tasks",
    broker="redis://localhost:6379/0",
    backend="redis://localhost:6379/0"
)

# 간단한 백그라운드 작업 (FastAPI BackgroundTasks)
def send_email_notification(email: str, message: str):
    # 이메일 발송 로직 시뮬레이션
    time.sleep(5)  # 5초 소요
    print(f"Email sent to {email}: {message}")

@app.post("/register/")
async def register_user(email: str, background_tasks: BackgroundTasks):
    # 사용자 등록 (빠른 작업)
    user_id = 123

    # 백그라운드에서 이메일 발송
    background_tasks.add_task(send_email_notification, email, "환영합니다!")

    # 즉시 응답 반환 (이메일 발송 기다리지 않음)
    return {"user_id": user_id, "message": "회원가입 성공"}

# Celery 작업 정의
@celery_app.task
def process_large_file(file_path: str):
    # 대용량 파일 처리 로직
    time.sleep(30)  # 30초 소요
    return f"File {file_path} processed successfully"

# 주기적 작업 (Celery Beat)
@celery_app.task
def sync_external_data():
    # 외부 API에서 데이터 동기화
    print("Syncing data from external API...")
    return "Sync completed"

@app.post("/upload/")
async def upload_file(file_path: str):
    # Celery 작업 큐에 추가
    task = process_large_file.delay(file_path)

    # 즉시 task_id 반환
    return {"task_id": task.id, "status": "processing"}

@app.get("/task/{task_id}")
async def get_task_status(task_id: str):
    # Celery 작업 상태 조회
    task = celery_app.AsyncResult(task_id)
    return {
        "task_id": task_id,
        "status": task.status,  # PENDING, STARTED, SUCCESS, FAILURE
        "result": task.result if task.ready() else None
    }

설명

이것이 하는 일: API 응답을 먼저 반환하고, 시간이 오래 걸리는 작업은 별도의 백그라운드 프로세스나 Celery 워커에서 비동기로 처리하여 응답 시간을 최소화합니다. 첫 번째로, BackgroundTasks는 FastAPI에 내장된 간단한 백그라운드 작업 메커니즘입니다.

background_tasks.add_task()로 함수와 인자를 등록하면, 응답이 클라이언트에게 전달된 후 해당 함수가 실행됩니다. send_email_notification 함수는 5초가 걸리지만, 사용자는 즉시 응답을 받고 이메일은 백그라운드에서 발송됩니다.

이것은 단일 프로세스 내에서 실행되므로 간단한 작업에 적합합니다. 그 다음으로, Celery는 분산 작업 큐 시스템으로, Redis나 RabbitMQ 같은 메시지 브로커를 통해 작업을 워커에게 전달합니다.

@celery_app.task 데코레이터로 작업을 정의하고, task.delay()로 작업을 큐에 추가하면, 별도의 Celery 워커 프로세스가 이를 가져와 실행합니다. 30초가 걸리는 파일 처리 작업도 API는 즉시 task_id를 반환하고, 워커가 백그라운드에서 처리합니다.

세 번째로, task_id를 통해 작업 상태를 조회할 수 있습니다. AsyncResult는 작업의 현재 상태(PENDING, STARTED, SUCCESS, FAILURE)와 결과를 반환하며, 프론트엔드에서 폴링이나 웹소켓으로 진행 상황을 사용자에게 보여줄 수 있습니다.

이것은 비동기 작업의 핵심으로, 사용자가 진행 상황을 알 수 있게 합니다. 마지막으로, Celery Beat를 사용하면 주기적인 작업(크론잡)을 스케줄링할 수 있습니다.

sync_external_data 같은 작업을 매일 자정에 실행하거나, 매 시간마다 실행하도록 설정할 수 있습니다. 여러분이 이 코드를 사용하면 무거운 작업이 API 성능에 영향을 주지 않으며, 실무에서는 여러 워커를 실행하여 동시에 수십 개의 작업을 병렬 처리할 수 있습니다.

이메일 발송, 리포트 생성, 데이터 동기화, 이미지 처리 등 다양한 시나리오에서 필수적입니다.

실전 팁

💡 BackgroundTasks는 앱이 종료되면 작업이 손실됩니다. 중요한 작업은 Celery를 사용하여 영속성을 보장하세요

💡 Celery 워커는 celery -A main.celery_app worker --loglevel=info 명령어로 별도 실행해야 합니다

💡 작업 재시도는 @celery_app.task(bind=True, max_retries=3)로 설정하여 일시적 실패에 대응하세요

💡 Flower(Celery 모니터링 도구)를 사용하면 작업 상태, 워커 상태, 실패한 작업을 웹 UI로 확인할 수 있습니다

💡 장시간 실행되는 작업은 self.update_state()로 중간 진행 상태를 업데이트하여 사용자에게 진행률을 보여주세요

#FastAPI#Router#Pydantic#AsyncIO#Dependency#Python