[Python] FastAPI예외 처리: 공식 문서 소개와 확장된 구현 방법

HTTP 예외 처리 개요

FastAPI는 클라이언트에게 오류를 알릴 때 HTTPException을 사용합니다. 예외 발생 시 HTTP 상태 코드와 함께 상세 메시지를 반환하며, 이는 클라이언트가 오류를 쉽게 이해할 수 있도록 돕습니다. 예외를 발생시키면 요청의 나머지 코드는 실행되지 않고 바로 오류 응답이 반환됩니다.

HTTP 예외 예제

다음은 HTTPException을 사용하여 요청한 리소스가 존재하지 않는 경우 404 오류를 반환하는 예제입니다.

from fastapi import FastAPI, HTTPException

app = FastAPI()

items = {"foo": "The Foo Wrestlers"}

@app.get("/items/{item_id}")
async def read_item(item_id: str):
    if item_id not in items:
        raise HTTPException(status_code=404, detail="Item not found")
    return {"item": items[item_id]}

• 404 상태 코드: 요청한 아이템이 존재하지 않을 때 404 오류와 함께 "Item not found"라는 메시지를 반환합니다.

커스텀 예외 처리기

FastAPI는 @app.exception_handler를 통해 사용자 정의 예외를 처리할 수 있는 핸들러를 제공합니다.

from fastapi import FastAPI, Request
from fastapi.responses import JSONResponse

class UnicornException(Exception):
    def __init__(self, name: str):
        self.name = name

app = FastAPI()

@app.exception_handler(UnicornException)
async def unicorn_exception_handler(request: Request, exc: UnicornException):
    return JSONResponse(
        status_code=418,
        content={"message": f"Oops! {exc.name} did something. There goes a rainbow..."},
    )

@app.get("/unicorns/{name}")
async def read_unicorn(name: str):
    if name == "yolo":
        raise UnicornException(name=name)
    return {"unicorn_name": name}

• 커스텀 예외: UnicornException을 정의하고, 특정 조건에서 이 예외를 발생시킵니다.
• 예외 처리기: unicorn_exception_handler를 통해 이 예외를 캐치하고 적절한 응답을 반환합니다.

확장된 예외 처리 구현

공식 문서의 방식은 기본적인 예외 처리를 다루지만, 좀 더 일관된 방식으로 예외 처리를 확장하는 것이 중요합니다. 이를 위해 예외를 중앙에서 관리하고, 클라이언트가 응답을 쉽게 이해할 수 있도록 일관된 응답 형식을 사용합니다.

예외 처리기 설계

단일 예외 처리기를 통해 모든 예외를 일관된 방식으로 처리할 수 있습니다. 이 핸들러는 다양한 HTTP 상태 코드와 데이터 검증 오류를 포함하여 모든 예외를 통합적으로 관리합니다.

# app/core/handlers/http_handlers.py

from fastapi import Request, HTTPException
from fastapi.responses import JSONResponse
from fastapi.exceptions import RequestValidationError
from app.schemas.response import ErrorResponse, ErrorDetail
from typing import Union, List

async def http_exception_handler(request: Request, exc: Union[HTTPException, RequestValidationError]):
    if isinstance(exc, HTTPException):
        if exc.status_code == 401:
            message = "인증이 필요합니다."
        elif exc.status_code == 402:
            message = "결제가 필요합니다."
        elif exc.status_code == 403:
            message = "접근이 금지되었습니다."
        elif exc.status_code == 404:
            message = "요청한 리소스를 찾을 수 없습니다."
        elif exc.status_code == 429:
            message = "요청이 너무 많습니다. 잠시 후 다시 시도해 주세요."
        else:
            message = str(exc.detail)

        response_content = ErrorResponse(
            status="error",
            code=0,
            message=message,
            errors=str(exc.status_code)
        )
        return JSONResponse(status_code=exc.status_code, content=response_content.model_dump_json())

    elif isinstance(exc, RequestValidationError):
        errors: List[ErrorDetail] = [
            ErrorDetail(loc=".".join(map(str, err["loc"])), msg=err["msg"], type=err["type"]) for err in exc.errors()
        ]
        response_content = ErrorResponse(
            status="error",
            code=0,
            message="유효하지 않은 요청입니다.",
            errors=errors
        )
        return JSONResponse(status_code=422, content=response_content.model_dump_json())

    response_content = ErrorResponse(
        status="error",
        code=0,
        message="서버에 오류가 발생했습니다.",
        errors=str(exc)
    )
    return JSONResponse(status_code=500, content=response_content.model_dump_json())

• HTTP 상태 코드 처리: 다양한 HTTP 상태 코드를 명확하게 처리합니다.
• 일관된 에러 응답: ErrorResponse 모델을 사용하여 모든 에러를 일관된 형식으로 응답합니다.

코드 예제: 사용자 CRUD API

확장된 예외 처리를 사용자 CRUD API에 적용하여, 모든 예외를 일관되게 처리합니다.

사용자 API 라우터

# app/routers/user.py

from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy.orm import Session
from app.db.session import get_db
from app.crud.user import get_user, create_user, update_user, delete_user, get_users
from app.schemas.user import UserCreate, User, UserUpdate
from app.schemas.response import SuccessResponse
from app.core.dependencies import get_http_exception_handler
from app.core.auth import verify_token  # JWT 인증 함수를 가져옵니다.
from typing import List

router = APIRouter()

@router.get("/", response_model=SuccessResponse, dependencies=[Depends(get_http_exception_handler), Depends(verify_token)])
def read_users(db: Session = Depends(get_db)):
    users = get_users(db=db)
    if not users:
        raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다.")
    return SuccessResponse(data=users, message="사용자 목록을 성공적으로 불러왔습니다.")

@router.post("/", response_model=SuccessResponse, dependencies=[Depends(get_http_exception_handler), Depends(verify_token)])
def create_user_endpoint(user: UserCreate, db: Session = Depends(get_db)):
    if not user.email:
        raise HTTPException(status_code=400, detail="이메일은 필수 항목입니다.")
    new_user = create_user(db=db, user=user)
    return SuccessResponse(data=new_user, message="사용자가 성공적으로 생성되었습니다.")

@router.get("/{user_id}", response_model=SuccessResponse, dependencies=[Depends(get_http_exception_handler), Depends(verify_token)])
def read_user(user_id: int, db: Session = Depends(get_db)):
    db_user = get_user(db=db, user_id=user_id)
    if db_user is None:
        raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다.")
    return SuccessResponse(data=db_user, message="사용자를 성공적으로 불러왔습니다.")

@router.put("/{user_id}", response_model=SuccessResponse, dependencies=[Depends(get_http_exception_handler), Depends(verify_token)])
def update_user_endpoint(user_id: int, user: UserUpdate, db: Session = Depends(get_db)):
    if user_id < 1:
        raise HTTPException(status_code=400, detail="유효하지 않은 사용자 ID입니다.")
    updated_user = update_user(db=db, user_id=user_id, user=user)
    return SuccessResponse(data=updated_user, message="사용자 정보가 성공적으로 업데이트되었습니다.")

@router.delete("/{user_id}", response_model=SuccessResponse, dependencies=[Depends(get_http_exception_handler), Depends(verify_token)])
def delete_user_endpoint(user_id: int, db: Session = Depends(get_db)):
    if user_id < 1:
        raise HTTPException(status_code=400, detail="유효하지 않은 사용자 ID입니다.")
    deleted_user = delete_user(db=db, user_id=user_id)
    if not deleted_user:
        raise HTTPException(status_code=404, detail="사용자를 삭제할 수 없습니다.")
    return SuccessResponse(data=None, message="사용자가 성공적으로 삭제되었습니다.")

• 일관된 예외 처리: 모든 예외는 http_exception_handler에 의해 처리되며, 클라이언트에게 일관된 응답 형식을 제공합니다.

메인 애플리케이션

# app/main.py

from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from app.api.v1.routers import router as api_v1_router
from app.db.session import create_tables
from app.core.config import settings
from app.core.dependencies import get_http_exception_handler
from fastapi.exceptions import RequestValidationError

app = FastAPI()

# CORS 설정
app.add_middleware(
    CORSMiddleware,
    allow_origins=settings.CORS_ORIGINS,
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 데이터베이스 테이블 생성
create_tables()

# API 라우터 포함
app.include_router(api_v1_router)

# 예외 처리기 등록
app.add_exception_handler(HTTPException, get_http_exception_handler())
app.add_exception_handler(RequestValidationError, get_http_exception_handler())

• CORS 설정: CORS 설정을 통해 특정 도메인에서만 접근을 허용합니다.
• 예외 처리기 등록: 모든 예외는 중앙에서 관리되어 일관된 방식으로 처리됩니다.

결론

FastAPI의 공식 문서에서는 기본적인 예외 처리 방법을 소개하지만, 이를 확장하여 일관된 예외 처리 시스템을 구축하는 것이 중요합니다. 이 글에서는 단일 예외 처리기를 통해 다양한 예외를 중앙에서 관리하고, 클라이언트에게 일관된 형식의 응답을 제공하는 방법을 설명했습니다. 이를 통해 웹 애플리케이션의 유지보수성을 높이고, 클라이언트 경험을 개선할 수 있습니다.

이제 여러분의 FastAPI 프로젝트에서도 일관된 예외 처리를 적용해 보세요. 이를 통해 예외 상황에서도 안정적이고 명확한 응답을 제공할 수 있을 것입니다.