FastAPI 비동기 실전: AsyncSQLAlchemy, BackgroundTasks, Dependency Injection으로 프로덕션 API 설계하기

비동기 Python, 제대로 쓰지 않으면 오히려 더 느리다

FastAPI는 2026년 현재 Python 백엔드 생태계에서 가장 빠르게 채택되는 웹 프레임워크입니다. async def를 라우트 핸들러에 붙이면 비동기 API가 완성된다는 인상이 있지만, 실제 프로덕션 환경에서는 그렇지 않습니다. 우리 팀이 일 평균 200만 건의 예약 요청을 처리하는 여행 플랫폼에서 FastAPI를 마이그레이션할 때, 잘못된 비동기 핸들러 때문에 오히려 p99 응답 시간이 40% 늘어난 적이 있습니다. 모든 라우트를 async def로 바꿨지만, 안에서 동기 SQLAlchemy를 그대로 호출하고 있었기 때문입니다.

이 글은 FastAPI의 비동기 모델이 내부적으로 어떻게 동작하는지, AsyncSQLAlchemy의 세션 생명주기를 어떻게 관리해야 하는지, BackgroundTasks와 Celery의 선택 기준, Pydantic v2 직렬화, 미들웨어에서의 Correlation ID 주입, 예외 처리 계층, Rate Limiting, 그리고 Uvicorn 배포 설정까지 프로덕션 운영에 필요한 내용을 코드와 함께 정리합니다.

---

1. FastAPI 비동기 모델의 진실: Starlette과 asyncio

FastAPI는 Starlette 위에 올라가 있습니다. Starlette은 ASGI(Asynchronous Server Gateway Interface) 애플리케이션 프레임워크로, Python의 asyncio 이벤트 루프를 직접 사용합니다.

Uvicorn이 새 HTTP 요청을 받으면 asyncio 이벤트 루프에 코루틴을 태스크로 등록합니다. 이벤트 루프는 단일 스레드에서 이 태스크들을 번갈아 실행합니다. await 지점에서 실행권을 양보하면 루프는 다른 태스크를 실행합니다. DB I/O, HTTP 호출, 파일 읽기를 await로 처리하면 그 시간 동안 다른 요청이 CPU를 사용할 수 있습니다.

핵심은 이벤트 루프가 단일 스레드라는 점입니다. CPU 바운드 연산(이미지 변환, 암호화, 무거운 계산)을 async 핸들러 안에서 직접 실행하면 이벤트 루프 전체가 블로킹됩니다. 다른 모든 요청이 그 연산이 끝날 때까지 기다립니다. 이런 작업은 반드시 asyncio.run_in_executor나 loop.run_in_executor로 스레드풀 또는 프로세스풀에 위임해야 합니다.

반대로 I/O 바운드 작업은 async로 처리할수록 처리량이 좋아집니다. 동기 방식으로 DB 쿼리 10개를 순차 실행하면 합산 대기 시간이 응답 시간이 됩니다. asyncio.gather로 병렬 실행하면 가장 느린 쿼리 하나의 대기 시간으로 줄어듭니다.

---

2. async def vs def: FastAPI가 동기 핸들러를 처리하는 방법

FastAPI는 라우트 핸들러를 async def와 def 두 가지 모두 지원합니다. 많은 개발자들이 def는 느리고 async def는 빠르다고 단순화하지만, 이것은 틀렸습니다. FastAPI 공식 문서는 이를 명확히 설명합니다.

def로 정의된 핸들러는 FastAPI가 외부 스레드풀(anyio가 관리하는 스레드풀)에서 실행합니다. 이벤트 루프를 블로킹하지 않습니다. 동기 SQLAlchemy나 동기 requests 라이브러리처럼 블로킹 코드가 있다면 def 핸들러가 오히려 안전합니다.

async def로 정의하면 핸들러는 이벤트 루프에서 직접 실행됩니다. 여기서 동기 블로킹 코드를 호출하면 이벤트 루프 전체가 멈춥니다. time.sleep(1)을 async def 안에서 호출하는 것이 대표적인 안티패턴입니다. await asyncio.sleep(1)이어야 합니다.

핸들러 유형	실행 위치	블로킹 코드 허용	권장 사용 사례
async def	이벤트 루프 직접 실행	불가	await로 감싼 async I/O 전용
def	anyio 스레드풀	가능	동기 라이브러리, 짧은 CPU 연산
async def + run_in_executor	스레드/프로세스풀	가능	CPU 집중 작업을 async에서 위임

실무에서 가장 흔한 실수는 ORM을 비동기로 전환하지 않은 채 라우트만 async def로 바꾸는 것입니다.

---

3. AsyncSession 생명주기와 트랜잭션 경계

SQLAlchemy 2.0의 비동기 지원은 AsyncEngine과 AsyncSession을 중심으로 동작합니다.

from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from typing import AsyncGenerator

DATABASE_URL = "postgresql+asyncpg://user:password@localhost:5432/appdb"

engine = create_async_engine(
    DATABASE_URL,
    pool_size=20,
    max_overflow=10,
    pool_timeout=30,
    pool_recycle=1800,
)

AsyncSessionLocal = async_sessionmaker(
    engine,
    class_=AsyncSession,
    expire_on_commit=False,
    autoflush=False,
)


async def get_db_session() -> AsyncGenerator[AsyncSession, None]:
    async with AsyncSessionLocal() as session:
        try:
            yield session
            await session.commit()
        except Exception:
            await session.rollback()
            raise
        finally:
            await session.close()

expire_on_commit=False는 중요한 설정입니다. 기본값인 True이면 commit() 이후 ORM 객체의 모든 속성이 만료되어, 응답 직렬화 시점에 Pydantic이 각 속성을 접근할 때마다 DB에 lazy load를 시도합니다. async 환경에서 lazy load는 MissingGreenlet 오류를 발생시킵니다.

---

4. Dependency Injection 설계: 함수 인자가 곧 의존성이다

FastAPI의 DI 시스템은 Depends()를 통해 동작합니다.

from fastapi import Depends, Header, HTTPException, status
from typing import Annotated

DbSession = Annotated[AsyncSession, Depends(get_db_session)]


async def get_current_user(
    authorization: Annotated[str | None, Header()] = None,
    db: DbSession = None,
) -> User:
    if not authorization or not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="인증 토큰이 없습니다")
    token = authorization.removeprefix("Bearer ")
    user = await UserService(db).get_by_token(token)
    if not user:
        raise HTTPException(status_code=401, detail="유효하지 않은 토큰입니다")
    return user


CurrentUser = Annotated[User, Depends(get_current_user)]


@router.post("/", response_model=OrderResponse)
async def create_order(
    payload: OrderCreateRequest,
    db: DbSession,
    current_user: CurrentUser,
    background_tasks: BackgroundTasks,
) -> OrderResponse:
    order = await OrderService(db).create(payload, user_id=current_user.id)
    background_tasks.add_task(
        NotificationService.send_order_confirmation,
        order_id=order.id,
        user_email=current_user.email,
    )
    return OrderResponse.model_validate(order)

Annotated 타입 별칭 패턴은 FastAPI 공식 문서에서도 권장합니다. 의존성 캐싱(Dependency Caching)도 이해해야 합니다. 같은 요청 내에서 Depends(get_db_session)을 여러 의존성이 동시에 선언해도 FastAPI는 기본적으로 세션을 한 번만 생성합니다.

---

5. BackgroundTasks vs Celery: 언제 어느 쪽인가

FastAPI의 BackgroundTasks는 HTTP 응답을 반환한 직후 같은 프로세스 안에서 코루틴을 실행합니다. 외부 큐, 브로커, 워커 프로세스가 필요 없습니다.

기준	BackgroundTasks	Celery + Redis/RabbitMQ
실행 위치	같은 프로세스	별도 워커 프로세스
외부 브로커	불필요	필요
재시도	직접 구현	기본 제공
작업 실패 시 손실	프로세스 종료 시 유실	큐에 보존
적합한 작업	경량 알림, 로그 기록	결제 정산, 이메일 발송

BackgroundTasks의 가장 큰 한계는 프로세스가 죽으면 태스크가 유실된다는 것입니다. 결제 정산이나 이메일 발송처럼 실패해서는 안 되는 작업은 Celery나 ARQ 같은 큐 기반 시스템을 사용해야 합니다.

---

6. Pydantic v2 직렬화 성능과 ORM 모드

Pydantic v2 공식 문서에 따르면 Rust 기반의 pydantic-core로 재작성되어 v1 대비 검증은 최대 17배, 직렬화는 최대 9배 빨라졌습니다.

from pydantic import BaseModel, ConfigDict, field_serializer
from datetime import datetime
from decimal import Decimal


class OrderResponse(BaseModel):
    model_config = ConfigDict(from_attributes=True)

    id: int
    status: str
    total_amount: Decimal
    created_at: datetime

    @field_serializer("total_amount")
    def serialize_price(self, value: Decimal) -> str:
        return f"{value:.2f}"


# 관계를 미리 로드해야 async 환경에서 lazy load 오류 방지
async def get_order_with_items(db: AsyncSession, order_id: int) -> Order | None:
    stmt = (
        select(Order)
        .where(Order.id == order_id)
        .options(selectinload(Order.items))
    )
    result = await db.execute(stmt)
    return result.scalar_one_or_none()

성능상 주의할 점은 관계(relationship) 접근입니다. async 환경에서 Pydantic이 직렬화 시점에 lazy load를 시도하면 MissingGreenlet이 발생합니다. 반드시 쿼리 시점에 selectinload 또는 joinedload로 관계를 eager load해야 합니다.

---

7. 미들웨어로 Correlation ID 주입하기

분산 시스템에서 하나의 사용자 요청이 여러 서비스를 거칠 때, 모든 로그를 연결하는 Correlation ID가 없으면 장애 추적이 어렵습니다. 구조적 로깅과 Correlation ID 전략에 대한 자세한 내용은 구조적 로깅 실전 가이드를 참고하십시오.

import uuid
from starlette.middleware.base import BaseHTTPMiddleware
from contextvars import ContextVar

correlation_id_var: ContextVar[str] = ContextVar("correlation_id", default="")


class CorrelationIdMiddleware(BaseHTTPMiddleware):
    HEADER_NAME = "X-Correlation-ID"

    async def dispatch(self, request, call_next):
        correlation_id = request.headers.get(self.HEADER_NAME) or str(uuid.uuid4())
        token = correlation_id_var.set(correlation_id)
        try:
            response = await call_next(request)
        finally:
            correlation_id_var.reset(token)
        response.headers[self.HEADER_NAME] = correlation_id
        return response

contextvars.ContextVar는 asyncio에서 요청 컨텍스트를 안전하게 전파하는 표준 방법입니다. 스레드 로컬과 달리 async 태스크 간에 컨텍스트가 섞이지 않습니다.

---

8. 예외 처리 계층: HTTPException, 커스텀 핸들러, 글로벌 hooks

FastAPI의 예외 처리는 세 계층으로 구성합니다. 첫째는 FastAPI 기본 HTTPException, 둘째는 도메인 예외 + @app.exception_handler, 셋째는 모든 예외를 잡는 글로벌 핸들러입니다.

도메인 예외를 HTTPException과 분리하는 것이 중요합니다. 비즈니스 로직 레이어에서 HTTP 상태 코드를 직접 알고 있으면 안 됩니다.

class DomainException(Exception):
    def __init__(self, message: str, code: str, context: dict | None = None):
        self.message = message
        self.code = code
        self.context = context or {}
        super().__init__(message)


class OrderNotFoundException(DomainException):
    def __init__(self, order_id: int):
        super().__init__(
            message=f"주문을 찾을 수 없습니다: {order_id}",
            code="ORDER_NOT_FOUND",
            context={"order_id": order_id},
        )


@app.exception_handler(OrderNotFoundException)
async def order_not_found_handler(request, exc):
    return JSONResponse(
        status_code=404,
        content={"error": exc.code, "message": exc.message, "context": exc.context},
    )


@app.exception_handler(Exception)
async def unhandled_exception_handler(request, exc):
    logger.exception("unhandled_exception", extra={"path": request.url.path})
    return JSONResponse(
        status_code=500,
        content={"error": "INTERNAL_ERROR", "message": "서버 내부 오류가 발생했습니다"},
    )

장애 전파를 막는 회복탄력성 패턴과 외부 서비스 호출 시 예외 처리 전략은 API 장애 전파를 막는 회복탄력성 패턴에서 더 자세히 다룹니다.

---

9. Rate Limiting과 동시성 제어

FastAPI 자체에는 Rate Limiting 기능이 내장되어 있지 않습니다. 프로덕션에서는 Nginx나 API 게이트웨이 수준에서 1차 Rate Limiting을 하고, 애플리케이션 레벨에서 2차로 더 세밀한 제어를 합니다.

동시성 제어도 중요한 주제입니다. 특정 리소스에 대한 동시 접근을 제한해야 하는 경우(예: 같은 재고에 대한 동시 주문), Redis의 분산 락이나 DB의 SELECT FOR UPDATE를 활용합니다.

from sqlalchemy import select


@router.post("/reserve/{product_id}")
async def reserve_product(
    product_id: int,
    quantity: int,
    db: DbSession,
    current_user: CurrentUser,
) -> dict:
    # 비관적 잠금: 같은 product_id에 대한 동시 요청을 직렬화
    stmt = (
        select(Product)
        .where(Product.id == product_id)
        .with_for_update()
    )
    result = await db.execute(stmt)
    product = result.scalar_one_or_none()

    if not product:
        raise HTTPException(status_code=404, detail="상품을 찾을 수 없습니다")
    if product.stock < quantity:
        raise HTTPException(status_code=409, detail=f"재고 부족")

    product.stock -= quantity
    await db.flush()

    return {"reserved": quantity, "remaining": product.stock}

---

10. Uvicorn worker 설정과 배포 체크리스트

Uvicorn 공식 설정 문서는 주요 환경 변수와 CLI 옵션을 상세히 설명합니다.

# 프로덕션 권장 실행 방식
gunicorn main:app \
  --worker-class uvicorn.workers.UvicornWorker \
  --workers 4 \
  --bind 0.0.0.0:8000 \
  --timeout 30 \
  --graceful-timeout 10 \
  --keep-alive 5 \
  --max-requests 1000 \
  --max-requests-jitter 100

# Uvicorn 단독 실행
uvicorn main:app \
  --host 0.0.0.0 --port 8000 --workers 4 \
  --loop uvloop --http httptools

--max-requests와 --max-requests-jitter 설정은 메모리 누수를 방어합니다. uvloop는 표준 asyncio 이벤트 루프를 libuv 기반의 고성능 구현체로 교체합니다. httptools는 기본 HTTP 파서인 h11을 더 빠른 구현체로 교체합니다.

컨테이너 환경에서는 gunicorn 없이 uvicorn --workers 단독 사용을 권장하는 경우도 있습니다. Kubernetes가 Pod 수로 워커를 스케일링하므로, 프로세스 내부에서 멀티 워커를 관리하는 복잡성을 줄일 수 있습니다.

---

결론

FastAPI는 올바르게 설계하면 Python 에코시스템에서 가장 생산성 높은 프로덕션 API 서버입니다. 하지만 비동기 모델의 동작 원리, ORM 세션 생명주기, DI 설계, 예외 처리 계층을 이해하지 못한 채 구현하면 오히려 더 많은 문제가 생깁니다.

다음 체크리스트로 설계를 검증하십시오.

• async def 핸들러 안에서 동기 블로킹 코드를 직접 호출하지 않는가
• AsyncSession에 expire_on_commit=False를 설정하고, 관계 로딩을 명시했는가
• BackgroundTasks로 등록한 작업이 프로세스 종료 시 유실되어도 무관한 경량 작업인가
• 예외 처리가 도메인 레이어와 HTTP 레이어를 분리하고 있는가
• Uvicorn 워커 수, --max-requests, uvloop, httptools 설정을 명시하고 부하 테스트로 검증했는가