FastAPI 실무 활용 완벽 가이드

Python 개발자를 위한 FastAPI 실무 활용 팁입니다. 의존성 주입, 비동기 처리, 미들웨어 활용 등 실전에서 꼭 필요한 핵심 패턴들을 다룹니다. 초급 개발자도 쉽게 따라할 수 있도록 친절하게 설명합니다.

---

목차

1. 의존성_주입
2. 비동기_처리
3. Pydantic_모델_검증
4. 미들웨어_활용
5. 백그라운드_태스크
6. 예외_처리
7. 라우터_분리
8. 데이터베이스_연동

---

1. 의존성 주입

시작하며

여러분이 API를 개발하다 보면 데이터베이스 연결, 인증 정보, 설정값 등을 여러 엔드포인트에서 반복해서 사용하게 됩니다. 매번 같은 코드를 복사-붙여넣기 하다가 나중에 수정할 때 모든 곳을 일일이 찾아다니며 고치느라 시간을 낭비한 경험, 있으시죠?

이런 문제는 프로젝트가 커질수록 더 심각해집니다. 코드 중복이 늘어나고, 테스트하기 어려워지며, 유지보수가 악몽이 됩니다.

특히 데이터베이스 세션 관리나 사용자 인증 같은 공통 로직은 모든 API 엔드포인트에서 필요한데, 이를 매번 작성하는 것은 비효율적입니다. 바로 이럴 때 필요한 것이 FastAPI의 의존성 주입(Dependency Injection)입니다.

한 번만 정의하면 모든 엔드포인트에서 재사용할 수 있고, 테스트할 때는 가짜 객체로 쉽게 교체할 수 있습니다.

개요

간단히 말해서, 의존성 주입은 함수나 클래스가 필요로 하는 객체를 외부에서 자동으로 제공받는 패턴입니다. FastAPI에서는 Depends() 함수를 사용해서 이를 구현합니다.

예를 들어, 데이터베이스 세션이 필요한 모든 API에서 똑같은 연결 로직을 작성하는 대신, 한 번만 정의하고 Depends(get_db)처럼 선언하면 FastAPI가 자동으로 세션을 생성해서 전달해줍니다. 기존에는 각 엔드포인트마다 db = SessionLocal()을 작성하고 try-finally로 세션을 닫아야 했다면, 이제는 의존성으로 정의한 함수가 모든 것을 처리해줍니다.

의존성 주입의 핵심 특징은 코드 재사용성, 테스트 용이성, 그리고 명확한 책임 분리입니다. 이러한 특징들은 프로젝트가 커질수록 더욱 빛을 발하며, 팀 협업 시에도 코드 이해도를 크게 높여줍니다.

코드 예제

from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session

app = FastAPI()

# 의존성 함수 정의
def get_db():
    db = SessionLocal()  # 데이터베이스 세션 생성
    try:
        yield db  # 세션을 엔드포인트에 전달
    finally:
        db.close()  # 요청 끝나면 자동으로 닫기

# 사용자 인증 의존성
def get_current_user(token: str, db: Session = Depends(get_db)):
    # 토큰 검증 후 사용자 정보 반환
    user = verify_token(token, db)
    return user

# 엔드포인트에서 의존성 주입 사용
@app.get("/users/me")
def read_user_me(current_user = Depends(get_current_user)):
    return current_user  # 자동으로 인증된 사용자 정보 전달됨

설명

이것이 하는 일: 의존성 주입은 각 API 엔드포인트가 필요로 하는 객체(데이터베이스 세션, 인증 정보 등)를 자동으로 생성하고 전달하며, 사용 후 정리까지 처리합니다. 첫 번째로, get_db() 함수가 의존성으로 정의됩니다.

이 함수는 yield 키워드를 사용하는데, 이것이 핵심입니다. yield 앞부분은 리소스를 생성하고, 뒷부분은 정리하는 역할을 합니다.

왜 이렇게 하냐면, API 요청이 처리되는 동안 세션을 유지하고, 응답을 보낸 후에 자동으로 닫아야 하기 때문입니다. 그 다음으로, get_current_user() 함수는 또 다른 의존성을 의존합니다.

즉, 의존성 안에서 다른 의존성을 사용할 수 있습니다. 이 함수가 실행되면서 먼저 get_db()를 호출해서 데이터베이스 세션을 받아오고, 그것을 사용해서 토큰을 검증합니다.

이런 식으로 의존성을 계층적으로 구성할 수 있어서 복잡한 로직도 깔끔하게 분리할 수 있습니다. 마지막으로, 실제 엔드포인트인 read_user_me()는 단순히 Depends(get_current_user)만 선언합니다.

FastAPI가 이것을 보고 자동으로 의존성 체인을 실행합니다. 즉, get_db() → get_current_user() → read_user_me() 순서로 호출되며, 최종적으로 인증된 사용자 정보가 엔드포인트에 전달됩니다.

여러분이 이 코드를 사용하면 데이터베이스 세션 관리를 걱정할 필요가 없고, 모든 엔드포인트에서 일관된 방식으로 인증을 처리할 수 있습니다. 테스트할 때는 app.dependency_overrides를 사용해서 실제 데이터베이스 대신 가짜 객체를 주입할 수 있어 단위 테스트가 매우 쉬워집니다.

또한 코드 중복이 사라지고, 나중에 인증 방식을 바꿀 때도 한 곳만 수정하면 되므로 유지보수 비용이 크게 줄어듭니다.

실전 팁

💡 의존성 함수는 일반 함수, 제너레이터, 클래스 모두 가능합니다. 복잡한 로직은 클래스로 만들면 상태를 유지하면서 재사용하기 좋습니다.

💡 yield 대신 return을 사용하면 정리 로직을 실행할 수 없으니, 리소스 관리가 필요한 경우 반드시 yield를 사용하세요.

💡 의존성은 캐싱됩니다. 같은 요청 내에서 동일한 의존성이 여러 번 호출되어도 한 번만 실행되므로 성능 걱정 없이 사용하세요.

💡 테스트 시 app.dependency_overrides[get_db] = mock_get_db처럼 오버라이드하면 실제 DB 없이도 테스트할 수 있습니다.

💡 의존성에서 예외를 발생시키면 전체 요청이 중단되므로, 인증 실패 같은 경우 HTTPException을 raise하여 명확한 에러 응답을 보내세요.

---

2. 비동기 처리

시작하며

여러분이 외부 API를 호출하거나 데이터베이스 쿼리를 실행할 때, 서버가 응답을 기다리는 동안 멈춰 있는 것을 본 적 있나요? 특히 사용자가 많아지면 서버가 하나의 요청을 처리하는 동안 다른 요청들이 대기하면서 전체 시스템이 느려지는 문제가 발생합니다.

이런 문제는 동기(Synchronous) 방식의 태생적 한계입니다. 전통적인 Flask나 Django에서는 한 요청이 끝나야 다음 요청을 처리할 수 있어서, I/O 작업이 많은 API에서는 성능 병목이 생깁니다.

결국 서버를 더 늘리거나 비싼 하드웨어를 사용해야 하는 상황에 처합니다. 바로 이럴 때 필요한 것이 FastAPI의 비동기 처리입니다.

async와 await 키워드를 사용하면 하나의 요청이 I/O를 기다리는 동안 다른 요청을 처리할 수 있어서, 같은 하드웨어로도 10배 이상의 성능 향상을 얻을 수 있습니다.

개요

간단히 말해서, 비동기 처리는 시간이 오래 걸리는 작업을 기다리는 동안 다른 일을 할 수 있게 해주는 프로그래밍 방식입니다. FastAPI는 Python의 asyncio를 기반으로 하며, async def로 함수를 정의하고 I/O 작업 앞에 await를 붙이기만 하면 됩니다.

예를 들어, 외부 API를 호출할 때 응답을 기다리는 1초 동안 서버는 다른 수백 개의 요청을 처리할 수 있습니다. 이것이 비동기의 힘입니다.

기존에는 한 번에 하나의 요청만 처리하던 것을 동시에 수천 개를 처리할 수 있게 됩니다. 멀티스레딩이나 멀티프로세싱보다 훨씬 가볍고 효율적이며, 컨텍스트 스위칭 오버헤드도 없습니다.

비동기 처리의 핵심 특징은 높은 동시성, 적은 리소스 사용, 그리고 빠른 응답 속도입니다. 특히 마이크로서비스 아키텍처에서 여러 서비스를 호출해야 할 때 그 진가를 발휘하며, 같은 서버로 더 많은 사용자를 처리할 수 있어 비용 절감 효과도 큽니다.

코드 예제

from fastapi import FastAPI
import httpx
import asyncio

app = FastAPI()

# 비동기 HTTP 클라이언트
async def fetch_user_data(user_id: int):
    async with httpx.AsyncClient() as client:
        # await로 응답을 기다리는 동안 다른 요청 처리
        response = await client.get(f"https://api.example.com/users/{user_id}")
        return response.json()

# 여러 API를 동시에 호출
@app.get("/dashboard/{user_id}")
async def get_dashboard(user_id: int):
    # 3개의 API를 병렬로 호출 - 순차적으로 하면 3초, 병렬로 하면 1초!
    user, posts, comments = await asyncio.gather(
        fetch_user_data(user_id),
        fetch_posts(user_id),
        fetch_comments(user_id)
    )
    return {"user": user, "posts": posts, "comments": comments}

설명

이것이 하는 일: 비동기 처리는 외부 API 호출, 데이터베이스 쿼리, 파일 읽기 같은 I/O 작업을 기다리는 동안 CPU를 놀리지 않고 다른 요청을 처리하여 서버의 처리량을 극대화합니다. 첫 번째로, async def로 함수를 정의합니다.

이것은 이 함수가 비동기 함수임을 의미하며, 내부에서 await를 사용할 수 있게 됩니다. httpx.AsyncClient()는 비동기 HTTP 클라이언트로, 일반적인 requests 라이브러리와 달리 await를 지원합니다.

왜 이렇게 하냐면, 네트워크 응답을 기다리는 동안 이벤트 루프에 제어권을 넘겨서 다른 작업을 할 수 있게 하기 위함입니다. 그 다음으로, asyncio.gather()를 사용해서 여러 비동기 작업을 동시에 실행합니다.

이것이 정말 강력한 부분인데, 만약 각 API 호출이 1초씩 걸린다면 순차적으로 실행하면 3초가 걸리지만, gather()를 사용하면 동시에 호출되어 1초 만에 끝납니다. 내부적으로는 각 API 호출이 시작되고, 응답을 기다리는 동안 다음 호출이 시작되며, 모든 응답이 돌아올 때까지 기다렸다가 한 번에 결과를 반환합니다.

마지막으로, FastAPI는 이 비동기 함수를 자동으로 이벤트 루프에서 실행합니다. 즉, 개발자는 async/await만 신경 쓰면 되고, 나머지 복잡한 이벤트 루프 관리는 FastAPI가 알아서 처리합니다.

요청이 들어오면 코루틴이 생성되고, await 지점에서 대기하며, I/O가 완료되면 다시 실행을 재개합니다. 여러분이 이 코드를 사용하면 외부 API 호출이 많은 시스템에서 극적인 성능 향상을 경험할 수 있습니다.

특히 마이크로서비스 환경에서 여러 서비스를 조합할 때 응답 시간을 획기적으로 줄일 수 있습니다. 서버 리소스도 훨씬 적게 사용하므로 클라우드 비용 절감 효과도 크며, 사용자 경험도 크게 개선됩니다.

단, 모든 I/O 라이브러리가 비동기를 지원하는 것은 아니므로, httpx, asyncpg, motor 같은 비동기 라이브러리를 사용해야 합니다.

실전 팁

💡 일반 requests 대신 httpx, psycopg2 대신 asyncpg, pymongo 대신 motor처럼 비동기를 지원하는 라이브러리를 사용하세요.

💡 CPU 집약적 작업(이미지 처리, 복잡한 계산)은 비동기로 해봤자 빨라지지 않습니다. 이런 경우 ProcessPoolExecutor를 사용하세요.

💡 비동기 함수 안에서 동기 함수를 호출하면 전체가 블로킹됩니다. 동기 함수는 await run_in_executor()로 감싸서 별도 쓰레드에서 실행하세요.

💡 데이터베이스 연결 풀 크기를 줄일 수 있습니다. 비동기는 적은 연결로도 많은 요청을 처리하므로 리소스 낭비를 막을 수 있습니다.

💡 asyncio.gather()에 return_exceptions=True를 추가하면 하나의 API가 실패해도 나머지는 계속 실행되어 부분적인 응답이라도 보낼 수 있습니다.

---

3. Pydantic 모델 검증

시작하며

여러분이 API를 개발할 때 클라이언트로부터 받은 데이터가 올바른지 검증하는 코드를 작성하느라 시간을 많이 쓰지 않나요? "이메일 형식이 맞나?", "나이가 0보다 큰가?", "필수 필드가 다 있나?" 같은 검증 로직을 일일이 작성하고, 에러 메시지까지 직접 만들어야 하는 번거로움이 있습니다.

이런 문제는 보안과 안정성에도 직결됩니다. 검증을 제대로 하지 않으면 잘못된 데이터가 데이터베이스에 저장되거나, SQL 인젝션 같은 공격에 노출될 수 있습니다.

또한 검증 코드가 여기저기 흩어져 있으면 나중에 규칙이 바뀔 때 모든 곳을 수정해야 하는 문제도 생깁니다. 바로 이럴 때 필요한 것이 Pydantic 모델 검증입니다.

데이터 구조를 클래스로 정의하기만 하면 FastAPI가 자동으로 검증하고, 타입 변환하고, 에러 메시지까지 생성해줍니다. 개발자는 비즈니스 로직에만 집중할 수 있습니다.

개요

간단히 말해서, Pydantic 모델은 Python 타입 힌트를 사용해서 데이터의 구조와 검증 규칙을 정의하는 클래스입니다. FastAPI는 Pydantic과 완벽하게 통합되어 있어서, 요청 바디나 쿼리 파라미터를 Pydantic 모델로 정의하면 자동으로 검증합니다.

예를 들어, 사용자 등록 API에서 이메일, 비밀번호, 나이를 받을 때 각 필드의 타입, 길이, 형식을 모델에 명시하면 FastAPI가 알아서 체크하고, 문제가 있으면 422 상태 코드와 함께 상세한 에러 메시지를 반환합니다. 기존에는 if not email or '@' not in email 같은 코드를 여러 줄 작성해야 했다면, 이제는 email: EmailStr 한 줄로 끝납니다.

타입도 자동으로 변환되어서 문자열 "123"을 보내면 정수 123으로 받을 수 있습니다. Pydantic의 핵심 특징은 자동 검증, 타입 변환, 명확한 에러 메시지, 그리고 IDE 자동완성 지원입니다.

이러한 특징들은 개발 속도를 높이고, 런타임 에러를 줄이며, API 문서도 자동으로 생성되게 해줍니다.

코드 예제

from fastapi import FastAPI
from pydantic import BaseModel, EmailStr, Field, validator
from typing import Optional

app = FastAPI()

# Pydantic 모델로 데이터 구조 정의
class UserCreate(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)  # 필수, 3-50자
    email: EmailStr  # 이메일 형식 자동 검증
    age: int = Field(..., ge=18, le=120)  # 18세 이상, 120세 이하
    password: str = Field(..., min_length=8)
    bio: Optional[str] = None  # 선택 필드

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

@app.post("/users")
def create_user(user: UserCreate):  # 자동으로 검증됨!
    return {"message": "사용자 생성 성공", "username": user.username}

설명

이것이 하는 일: Pydantic 모델은 클라이언트로부터 받은 데이터를 파싱하고, 타입을 확인하고, 제약 조건을 검증한 후, 검증에 실패하면 자동으로 상세한 에러 응답을 생성합니다. 첫 번째로, BaseModel을 상속받아 클래스를 정의합니다.

각 필드는 Python 타입 힌트로 타입을 지정하고, Field()로 추가 제약을 정의합니다. 예를 들어, min_length=3은 최소 길이, ge=18은 "greater than or equal"로 18 이상을 의미합니다.

EmailStr은 Pydantic이 제공하는 특수 타입으로, 이메일 형식을 자동으로 검증해줍니다. 왜 이렇게 하냐면, 코드만 봐도 어떤 데이터를 받는지 명확히 알 수 있고, 런타임에 자동으로 검증되기 때문입니다.

그 다음으로, @validator 데코레이터로 커스텀 검증 로직을 추가할 수 있습니다. 이것은 내장 검증만으로 부족할 때 사용하며, 예를 들어 비밀번호에 숫자가 포함되어야 한다는 복잡한 규칙을 구현할 수 있습니다.

validator는 필드 값이 기본 검증을 통과한 후에 실행되며, ValueError를 raise하면 FastAPI가 이를 422 에러로 변환해서 클라이언트에게 전달합니다. 마지막으로, 엔드포인트 함수의 파라미터로 Pydantic 모델을 선언하면 FastAPI가 마법을 부립니다.

요청이 들어오면 JSON 바디를 파싱하고, 각 필드를 검증하고, 타입을 변환한 후, 모든 것이 통과하면 UserCreate 인스턴스를 생성해서 함수에 전달합니다. 개발자는 이미 검증된 데이터를 안심하고 사용할 수 있습니다.

여러분이 이 코드를 사용하면 검증 코드를 한 곳에 모을 수 있어서 유지보수가 쉬워지고, API 문서(Swagger)에도 자동으로 필드 설명이 표시되어 프론트엔드 개발자와의 협업이 원활해집니다. IDE에서 자동완성도 지원되어 user. 만 입력해도 사용 가능한 필드를 볼 수 있습니다.

또한 잘못된 데이터가 시스템에 들어올 가능성이 거의 없어져서 보안과 안정성이 크게 향상되며, 에러 메시지도 자동으로 생성되어 클라이언트가 무엇이 잘못되었는지 명확히 알 수 있습니다.

실전 팁

💡 EmailStr, HttpUrl, UUID 같은 Pydantic의 특수 타입을 활용하면 복잡한 검증 로직을 한 줄로 처리할 수 있습니다.

💡 Config 내부 클래스로 orm_mode = True를 설정하면 SQLAlchemy 모델을 Pydantic 모델로 자동 변환할 수 있어 편리합니다.

💡 민감한 정보(비밀번호)는 응답에 포함되지 않도록 별도의 응답 모델을 만들거나 Field(exclude=True)를 사용하세요.

💡 @validator의 pre=True 옵션을 사용하면 타입 변환 전에 검증할 수 있어, 예를 들어 문자열을 소문자로 변환한 후 저장할 수 있습니다.

💡 공통 필드가 많으면 베이스 모델을 만들어 상속하면 중복을 줄일 수 있습니다. 예: UserBase, UserCreate(UserBase), UserUpdate(UserBase).

---

4. 미들웨어 활용

시작하며

여러분이 모든 API 요청에 대해 로깅을 하거나, 응답 시간을 측정하거나, CORS 헤더를 추가하거나, 인증을 체크해야 할 때 각 엔드포인트마다 같은 코드를 반복하고 있지는 않나요? 이런 공통 로직을 매번 작성하면 코드가 지저분해지고, 하나라도 빼먹으면 보안 구멍이 생깁니다.

이런 문제는 횡단 관심사(Cross-cutting Concerns)라고 불리며, 비즈니스 로직과는 별개로 모든 요청에 적용되어야 하는 기능들입니다. 로깅, 모니터링, 보안, 에러 처리 같은 것들이 여기 해당하며, 이를 각 엔드포인트에 흩어 놓으면 관리가 불가능해집니다.

바로 이럴 때 필요한 것이 FastAPI의 미들웨어입니다. 미들웨어는 모든 요청이 엔드포인트에 도달하기 전과 응답이 클라이언트로 가기 전에 실행되는 코드로, 공통 로직을 한 곳에서 중앙 집중식으로 관리할 수 있게 해줍니다.

개요

간단히 말해서, 미들웨어는 요청-응답 사이클의 중간에서 실행되는 함수로, 모든 API 호출에 공통으로 적용할 로직을 정의하는 곳입니다. FastAPI에서는 @app.middleware("http") 데코레이터로 미들웨어를 정의하며, 요청 객체를 받아서 처리한 후 다음 핸들러를 호출하고, 응답을 받아서 수정할 수 있습니다.

예를 들어, 요청이 들어오면 시작 시간을 기록하고, 응답이 나갈 때 소요 시간을 계산해서 헤더에 추가하는 식입니다. 기존에는 각 엔드포인트에서 start_time = time.time() 같은 코드를 작성했다면, 이제는 미들웨어 한 곳에서 정의하면 모든 엔드포인트에 자동 적용됩니다.

새로운 엔드포인트를 추가해도 별도 작업이 필요 없습니다. 미들웨어의 핵심 특징은 중앙 집중식 관리, 자동 적용, 그리고 요청-응답 전체 제어입니다.

이러한 특징들은 보안 정책을 일관되게 적용하고, 모니터링을 체계적으로 구축하며, 코드 중복을 제거하는 데 필수적입니다.

코드 예제

from fastapi import FastAPI, Request
import time
import logging

app = FastAPI()
logger = logging.getLogger(__name__)

# 미들웨어 정의
@app.middleware("http")
async def add_process_time_header(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
    response.headers["X-Process-Time"] = str(process_time)
    logger.info(f"요청 완료: {process_time:.3f}초")

    return response  # 수정된 응답 반환

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    return {"item_id": item_id}  # 미들웨어가 자동 적용됨

설명

이것이 하는 일: 미들웨어는 모든 HTTP 요청을 가로채서 전처리를 수행하고, 엔드포인트를 실행한 후, 응답을 후처리해서 클라이언트에게 보내는 파이프라인을 구성합니다. 첫 번째로, @app.middleware("http") 데코레이터로 함수를 미들웨어로 등록합니다.

이 함수는 request와 call_next 두 개의 파라미터를 받는데, request는 클라이언트의 요청 객체이고, call_next는 다음 단계(다른 미들웨어나 최종 엔드포인트)를 호출하는 함수입니다. await call_next(request) 전의 코드는 요청 전처리, 후의 코드는 응답 후처리를 담당합니다.

왜 이렇게 하냐면, 요청이 들어올 때와 나갈 때 각각 다른 작업을 할 수 있게 하기 위함입니다. 그 다음으로, 요청 전처리 부분에서 start_time을 기록하고 로그를 남깁니다.

이후 await call_next(request)를 호출하면 FastAPI가 적절한 엔드포인트를 찾아 실행하고, 그 결과를 response 객체로 반환합니다. 이것이 핵심인데, 미들웨어는 엔드포인트가 무엇을 하는지 몰라도 되며, 단지 요청과 응답을 중계하면서 원하는 작업을 수행합니다.

마지막으로, 응답 후처리 부분에서 경과 시간을 계산하고 response.headers에 추가합니다. 이렇게 하면 클라이언트는 응답 헤더에서 X-Process-Time을 보고 API가 얼마나 빨리 응답했는지 알 수 있습니다.

수정된 응답을 return하면 이것이 클라이언트에게 전달됩니다. 만약 여러 미들웨어가 있다면 체인처럼 연결되어 순차적으로 실행됩니다.

여러분이 이 코드를 사용하면 성능 모니터링을 모든 API에 일괄 적용할 수 있고, 느린 엔드포인트를 쉽게 찾아낼 수 있습니다. 로그도 일관된 형식으로 남아서 문제 발생 시 추적이 쉬워지며, 나중에 인증이나 CORS 같은 기능도 미들웨어로 추가하면 됩니다.

또한 미들웨어는 예외를 캐치할 수도 있어서 전역 에러 핸들러로도 사용할 수 있습니다. 코드가 한 곳에 모여 있어 팀원들도 공통 로직을 쉽게 이해할 수 있습니다.

실전 팁

💡 여러 미들웨어는 등록 순서대로 실행됩니다. 인증 체크는 가장 먼저, 로깅은 가장 나중에 등록하는 것이 일반적입니다.

💡 미들웨어에서 예외를 처리하려면 try-except로 call_next를 감싸면 모든 엔드포인트의 에러를 한 곳에서 처리할 수 있습니다.

💡 CORS는 CORSMiddleware를 사용하세요. app.add_middleware(CORSMiddleware, allow_origins=["*"])처럼 간단히 추가할 수 있습니다.

💡 무거운 작업(DB 쿼리, 외부 API 호출)은 미들웨어에서 피하세요. 모든 요청에 적용되므로 성능 병목이 될 수 있습니다.

💡 특정 경로만 제외하려면 미들웨어 안에서 if request.url.path.startswith("/health") 같은 조건문으로 분기 처리하세요.

---

5. 백그라운드 태스크

시작하며

여러분이 회원가입 후 환영 이메일을 보내거나, 이미지를 업로드한 후 썸네일을 생성하거나, 로그를 파일에 기록해야 할 때 이런 작업들이 끝날 때까지 사용자를 기다리게 하고 있지는 않나요? 이메일 발송에 3초가 걸리면 API 응답도 3초가 걸려서 사용자 경험이 나빠지고, 타임아웃 에러도 발생하기 쉽습니다.

이런 문제는 즉시 응답해야 하는 API의 특성과 시간이 걸리는 작업의 충돌에서 발생합니다. 사용자는 회원가입이 성공했는지만 빨리 알고 싶은데, 이메일이 실제로 발송될 때까지 기다려야 하는 것은 비효율적입니다.

또한 작업 도중 에러가 나면 전체 요청이 실패하는 문제도 있습니다. 바로 이럴 때 필요한 것이 FastAPI의 백그라운드 태스크입니다.

응답을 먼저 보내고, 시간이 걸리는 작업은 백그라운드에서 비동기로 처리하여 사용자는 즉시 결과를 받을 수 있고, 서버는 효율적으로 리소스를 사용할 수 있습니다.

개요

간단히 말해서, 백그라운드 태스크는 API 응답을 보낸 후에 백그라운드에서 실행되는 함수로, 사용자가 기다릴 필요 없는 작업을 처리하는 데 사용됩니다. FastAPI의 BackgroundTasks 객체를 사용하면 함수와 파라미터를 등록하기만 하면 되고, 응답이 클라이언트에게 전송된 직후 자동으로 실행됩니다.

예를 들어, 사용자 등록 API에서 DB에 저장한 후 바로 200 응답을 보내고, 이메일 발송은 백그라운드에서 처리하는 식입니다. 기존에는 Celery 같은 무거운 작업 큐를 설정해야 했다면, 이제는 간단한 작업은 FastAPI 내장 기능만으로 충분합니다.

별도의 브로커(Redis, RabbitMQ)도 필요 없고, 설정도 필요 없습니다. 백그라운드 태스크의 핵심 특징은 빠른 응답 속도, 간단한 구현, 그리고 에러 격리입니다.

단, 서버가 재시작되면 실행 중인 태스크가 사라지므로, 중요한 작업이나 긴 작업은 Celery 같은 전문 작업 큐를 사용해야 합니다.

코드 예제

from fastapi import FastAPI, BackgroundTasks
import time

app = FastAPI()

# 백그라운드에서 실행될 함수
def send_welcome_email(email: str, username: str):
    # 이메일 발송에 3초 걸림
    time.sleep(3)
    print(f"{username}님께 {email}로 환영 이메일 발송 완료")

def log_user_activity(user_id: int, action: str):
    # 로그 파일에 기록
    with open("activity.log", "a") as f:
        f.write(f"{time.time()}: User {user_id} - {action}\n")

# 엔드포인트에서 백그라운드 태스크 등록
@app.post("/signup")
async def signup(email: str, username: str, background_tasks: BackgroundTasks):
    # 1. 먼저 DB에 사용자 저장 (핵심 작업)
    user_id = save_user_to_db(email, username)

    # 2. 백그라운드 태스크 등록 (시간 걸리는 작업)
    background_tasks.add_task(send_welcome_email, email, username)
    background_tasks.add_task(log_user_activity, user_id, "signup")

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

설명

이것이 하는 일: 백그라운드 태스크는 중요한 비즈니스 로직은 먼저 처리하고 즉시 응답을 보낸 후, 부가적인 작업은 응답과 별개로 비동기적으로 실행하여 사용자 경험을 개선합니다. 첫 번째로, 백그라운드에서 실행할 함수를 일반 Python 함수로 정의합니다.

이 함수는 async def일 필요는 없으며, 일반 함수도 가능합니다. send_welcome_email() 함수는 이메일 발송 로직을 담고 있으며, 실제로는 SMTP 서버에 연결해서 이메일을 보내는 코드가 들어갑니다.

왜 이것을 백그라운드로 하냐면, 이메일 발송 성공 여부가 회원가입 API의 성공 여부와 무관하기 때문입니다. 사용자는 가입이 됐다는 것만 알면 되고, 이메일은 나중에 받아도 됩니다.

그 다음으로, 엔드포인트 함수의 파라미터에 BackgroundTasks 객체를 선언합니다. FastAPI가 자동으로 이 객체를 생성해서 전달하며, background_tasks.add_task(함수, 인자1, 인자2, ...)로 태스크를 등록합니다.

여러 개의 태스크를 등록할 수 있고, 등록 순서대로 실행됩니다. 내부적으로는 큐에 추가되며, 응답이 클라이언트에게 전송된 직후 FastAPI가 순차적으로 실행합니다.

마지막으로, return 문으로 응답을 보내는 순간 클라이언트는 즉시 응답을 받습니다. 이메일 발송이나 로그 기록은 아직 시작도 안 했지만, 사용자는 이미 "회원가입 성공" 메시지를 보고 있습니다.

그 사이 서버는 백그라운드에서 등록된 태스크들을 차례로 실행하며, 에러가 나도 이미 응답을 보낸 후라 사용자에게는 영향이 없습니다. 여러분이 이 코드를 사용하면 API 응답 시간이 극적으로 줄어듭니다.

3초 걸리던 API가 0.1초로 단축되어 사용자는 즉각적인 피드백을 받으며, 서버는 백그라운드에서 천천히 작업을 처리합니다. 또한 이메일 발송이 실패해도 회원가입은 성공하므로 사용자 경험이 개선되고, 나중에 실패한 이메일만 재시도하면 됩니다.

단, 결제 처리나 중요한 비즈니스 로직은 백그라운드 태스크로 하면 안 되며, 반드시 응답 전에 완료해야 합니다. 또한 서버가 재시작되면 실행 중인 태스크가 손실되므로, 중요한 작업은 Celery + Redis 조합을 사용하는 것이 안전합니다.

실전 팁

💡 백그라운드 태스크는 간단한 작업(이메일, 로그, 알림)에만 사용하고, 오래 걸리는 작업(동영상 인코딩, 대량 데이터 처리)은 Celery를 사용하세요.

💡 태스크 함수에서 예외가 발생하면 FastAPI가 로그에 기록하지만 재시도하지 않으므로, 중요한 작업은 try-except로 감싸고 직접 재시도 로직을 구현하세요.

💡 의존성 주입(Depends)과 함께 사용할 수 있습니다. 예: background_tasks.add_task(log_activity, db=Depends(get_db)) 형태로 DB 세션 전달 가능.

💡 파일 업로드 후 처리(이미지 리사이징, PDF 생성)는 백그라운드 태스크의 완벽한 사용 사례입니다. 파일만 저장하고 바로 응답하세요.

💡 여러 태스크가 같은 리소스(DB, 파일)에 접근하면 경합이 발생할 수 있으니, 잠금(Lock)이나 트랜잭션 격리를 고려하세요.

---

6. 예외 처리

시작하며

여러분이 API를 개발하다 보면 데이터베이스 연결 실패, 잘못된 입력값, 권한 없음, 리소스 없음 같은 다양한 에러 상황이 발생합니다. 이런 에러들을 제대로 처리하지 않으면 서버가 500 에러를 내뱉으며, 클라이언트는 무엇이 잘못됐는지 알 수 없어 답답해합니다.

이런 문제는 사용자 경험과 디버깅 모두에 악영향을 미칩니다. 막연한 "Internal Server Error" 메시지는 프론트엔드 개발자에게도, 사용자에게도 도움이 안 되며, 로그를 뒤져봐야 원인을 알 수 있습니다.

또한 보안상 민감한 정보가 에러 메시지에 노출될 위험도 있습니다. 바로 이럴 때 필요한 것이 FastAPI의 체계적인 예외 처리입니다.

HTTPException으로 명확한 상태 코드와 메시지를 반환하고, 커스텀 예외 핸들러로 전역적으로 에러를 관리하며, 사용자에게는 친절한 메시지를, 개발자에게는 상세한 로그를 제공할 수 있습니다.

개요

간단히 말해서, 예외 처리는 예상 가능한 에러 상황을 명시적으로 다루어 적절한 HTTP 상태 코드와 메시지를 클라이언트에 반환하는 메커니즘입니다. FastAPI는 HTTPException을 제공하며, 이를 raise하면 지정한 상태 코드와 메시지가 JSON 형태로 응답됩니다.

예를 들어, 사용자를 찾지 못하면 HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")를 raise하면 됩니다. 또한 @app.exception_handler로 특정 예외를 전역적으로 처리할 수도 있습니다.

기존에는 각 엔드포인트에서 if not user: return JSONResponse(...)처럼 직접 응답을 만들어야 했다면, 이제는 예외를 던지기만 하면 FastAPI가 알아서 적절한 응답을 생성합니다. 코드도 깔끔해지고, 에러 처리 로직도 중앙화됩니다.

예외 처리의 핵심 특징은 명확한 상태 코드, 일관된 에러 응답 형식, 그리고 전역 에러 핸들링입니다. 이러한 특징들은 API 사용자가 에러를 쉽게 이해하고 대응할 수 있게 하며, 개발자는 디버깅과 모니터링을 체계적으로 할 수 있게 해줍니다.

코드 예제

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

app = FastAPI()

# 커스텀 예외 클래스
class UserNotFoundException(Exception):
    def __init__(self, user_id: int):
        self.user_id = user_id

# 전역 예외 핸들러
@app.exception_handler(UserNotFoundException)
async def user_not_found_handler(request: Request, exc: UserNotFoundException):
    return JSONResponse(
        status_code=404,
        content={"error": "사용자 없음", "user_id": exc.user_id}
    )

# 엔드포인트에서 예외 사용
@app.get("/users/{user_id}")
async def get_user(user_id: int):
    user = get_user_from_db(user_id)

    # 일반 HTTPException
    if not user:
        raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")

    # 권한 체크
    if not user.is_active:
        raise HTTPException(status_code=403, detail="비활성화된 계정입니다")

    return user

@app.get("/items/{item_id}")
async def get_item(item_id: int):
    item = get_item_from_db(item_id)

    # 커스텀 예외
    if not item.owner:
        raise UserNotFoundException(user_id=item.owner_id)

    return item

설명

이것이 하는 일: 예외 처리는 비즈니스 로직 중 발생하는 에러 상황을 감지하고, 적절한 HTTP 상태 코드와 사용자 친화적인 메시지를 생성하여 클라이언트에게 전달하며, 서버 로그에는 상세한 정보를 기록합니다. 첫 번째로, HTTPException을 raise하는 것이 가장 기본적인 방법입니다.

status_code에는 HTTP 상태 코드(200, 404, 500 등)를, detail에는 에러 메시지를 넣습니다. FastAPI는 이 예외를 자동으로 캐치해서 JSON 응답 {"detail": "사용자를 찾을 수 없습니다"}로 변환합니다.

왜 예외를 사용하냐면, 함수 중간에서 에러가 발생했을 때 즉시 실행을 중단하고 상위로 전파할 수 있기 때문입니다. 여러 단계의 함수 호출이 있어도 예외는 자동으로 최상위까지 올라갑니다.

그 다음으로, 커스텀 예외 클래스를 정의하면 더 구조화된 에러 처리가 가능합니다. UserNotFoundException은 일반 Python 예외를 상속받으며, 필요한 정보(user_id)를 속성으로 저장합니다.

이렇게 하면 에러 종류별로 다른 로직을 적용할 수 있습니다. 예를 들어, 사용자 없음은 404를, 권한 없음은 403을 반환하는 식으로 세밀하게 제어할 수 있습니다.

마지막으로, @app.exception_handler 데코레이터로 전역 예외 핸들러를 등록합니다. 이것은 특정 예외가 발생했을 때 자동으로 호출되며, 원하는 형태의 응답을 만들어 반환할 수 있습니다.

모든 엔드포인트에서 발생한 UserNotFoundException은 이 핸들러가 처리하므로, 각 엔드포인트에서는 그냥 예외를 던지기만 하면 됩니다. 또한 여기서 로깅, 모니터링 알림, Sentry 전송 같은 공통 로직도 추가할 수 있습니다.

여러분이 이 코드를 사용하면 에러 응답이 일관된 형식으로 통일되어 프론트엔드에서 처리하기 쉬워지고, 사용자에게도 명확한 메시지를 전달할 수 있습니다. 또한 예외 핸들러에서 로그를 남기면 어떤 에러가 얼마나 자주 발생하는지 추적할 수 있어 서비스 품질 개선에 도움이 됩니다.

비즈니스 로직도 깔끔해지는데, if not user: return {...}처럼 응답을 직접 만드는 대신 raise HTTPException(...)으로 간결하게 작성할 수 있습니다. 보안상으로도 중요한데, 프로덕션 환경에서는 민감한 정보(DB 연결 문자열, 스택 트레이스)를 숨기고 일반적인 메시지만 보여줄 수 있습니다.

실전 팁

💡 상태 코드를 정확히 사용하세요. 404는 리소스 없음, 403은 권한 없음, 401은 인증 필요, 422는 검증 실패, 500은 서버 에러입니다.

💡 HTTPException의 headers 파라미터로 추가 헤더를 설정할 수 있습니다. 예: 인증 실패 시 headers={"WWW-Authenticate": "Bearer"} 추가.

💡 Pydantic ValidationError는 FastAPI가 자동으로 422 에러로 변환하지만, 메시지를 커스터마이징하려면 @app.exception_handler(RequestValidationError)를 사용하세요.

💡 프로덕션에서는 HTTPException(status_code=500) 대신 일반 예외를 로깅하고 "서버 오류가 발생했습니다" 같은 일반 메시지를 반환하세요.

💡 Sentry나 CloudWatch 같은 모니터링 도구와 통합하려면 예외 핸들러에서 해당 라이브러리의 API를 호출하여 에러를 자동으로 보고하세요.

---

7. 라우터 분리

시작하며

여러분이 프로젝트가 커지면서 main.py에 수십 개의 엔드포인트가 몰려 있어서 코드를 찾기 어렵고, 팀원들과 작업할 때 충돌이 자주 발생하지는 않나요? 사용자 관련 API, 상품 관련 API, 주문 관련 API가 한 파일에 섞여 있으면 어디에 무엇이 있는지 파악하기 힘들고, 유지보수도 악몽이 됩니다.

이런 문제는 모놀리틱 구조의 태생적 한계입니다. 하나의 파일에 모든 것을 넣으면 처음에는 간단해 보이지만, 기능이 추가될수록 파일이 수천 줄로 늘어나고, 책임이 불명확해지며, 테스트도 어려워집니다.

특히 팀 협업 시 모두가 같은 파일을 수정하므로 Git 충돌이 빈번하게 발생합니다. 바로 이럴 때 필요한 것이 FastAPI의 라우터 분리입니다.

기능별로 파일을 나누고, 각 파일은 APIRouter로 엔드포인트를 정의한 후, 메인 앱에 통합하는 방식으로 코드를 모듈화할 수 있습니다. 이렇게 하면 코드 구조가 명확해지고, 각 모듈을 독립적으로 개발하고 테스트할 수 있습니다.

개요

간단히 말해서, 라우터 분리는 관련된 엔드포인트들을 별도의 파일로 분리하여 모듈화하고, 메인 앱에서 이들을 조합하는 설계 패턴입니다. FastAPI의 APIRouter는 앱의 미니 버전처럼 작동하며, 라우터마다 독립적인 엔드포인트를 정의할 수 있습니다.

예를 들어, users.py에서 사용자 관련 API를, products.py에서 상품 관련 API를 정의하고, main.py에서 app.include_router(users_router, prefix="/users")처럼 통합합니다. 이렇게 하면 /users/me, /users/login 같은 엔드포인트가 자동으로 생성됩니다.

기존에는 모든 엔드포인트를 @app.get(...)으로 정의했다면, 이제는 @router.get(...)으로 정의하고 나중에 조합합니다. 각 라우터는 독립적인 의존성, 태그, 미들웨어를 가질 수 있어서 고도로 모듈화된 구조를 만들 수 있습니다.

라우터 분리의 핵심 특징은 코드 모듈화, 책임 분리, 그리고 재사용성입니다. 이러한 특징들은 대규모 프로젝트에서 필수적이며, 팀 협업을 원활하게 하고, 새로운 기능을 추가할 때 기존 코드에 영향을 주지 않는 확장 가능한 구조를 만들어줍니다.

코드 예제

# users.py - 사용자 관련 라우터
from fastapi import APIRouter, Depends

router = APIRouter(
    prefix="/users",  # 모든 경로 앞에 /users 추가
    tags=["users"],   # Swagger에서 그룹화
)

@router.get("/me")
async def get_current_user(user = Depends(get_current_user)):
    return user

@router.post("/login")
async def login(credentials: LoginRequest):
    return {"access_token": "..."}

# products.py - 상품 관련 라우터
from fastapi import APIRouter

router = APIRouter(prefix="/products", tags=["products"])

@router.get("/")
async def list_products():
    return {"products": [...]}

# main.py - 메인 앱에서 라우터 통합
from fastapi import FastAPI
from app.routers import users, products

app = FastAPI()

# 라우터 등록
app.include_router(users.router)
app.include_router(products.router)

# 결과: /users/me, /users/login, /products/ 엔드포인트 생성됨

설명

이것이 하는 일: 라우터 분리는 거대한 단일 파일을 작고 집중된 모듈들로 나누어 각 모듈이 하나의 도메인(사용자, 상품, 주문 등)만 담당하게 하고, 메인 앱은 이들을 조합하는 역할만 수행하게 합니다. 첫 번째로, 각 도메인별로 파일을 만들고 APIRouter()를 생성합니다.

prefix 파라미터는 이 라우터의 모든 경로 앞에 붙는 접두사를 정의하며, 예를 들어 prefix="/users"로 설정하면 @router.get("/me")는 자동으로 /users/me가 됩니다. tags는 Swagger 문서에서 엔드포인트를 그룹화하는 데 사용되며, 같은 태그를 가진 엔드포인트들이 함께 표시됩니다.

왜 이렇게 하냐면, 각 모듈이 독립적으로 작동하면서도 일관된 URL 구조를 유지할 수 있기 때문입니다. 그 다음으로, 각 라우터에서는 @router.get, @router.post 같은 데코레이터를 사용합니다.

이것은 @app.get과 동일하게 작동하지만, 라우터 인스턴스에 등록됩니다. 라우터는 앱과 독립적이므로, 같은 라우터를 여러 앱에서 재사용하거나, 버전별로 다른 설정을 적용할 수도 있습니다.

또한 각 라우터는 자체 의존성을 가질 수 있어서, 예를 들어 관리자 라우터는 모든 엔드포인트에 관리자 권한 체크를 자동으로 적용할 수 있습니다. 마지막으로, main.py에서 app.include_router(users.router)로 라우터를 메인 앱에 통합합니다.

이 순간 라우터에 정의된 모든 엔드포인트가 앱에 추가되며, FastAPI는 이들을 하나의 통합된 API로 취급합니다. 여러 라우터를 추가할 수 있고, 각 라우터는 독립적으로 작동하므로 충돌 걱정이 없습니다.

또한 라우터를 조건부로 포함할 수도 있어서, 예를 들어 개발 환경에서만 디버깅 라우터를 활성화할 수 있습니다. 여러분이 이 코드를 사용하면 프로젝트 구조가 명확해져서 새로운 팀원도 쉽게 코드를 이해할 수 있고, 특정 기능만 수정할 때 해당 파일만 열면 되므로 생산성이 크게 향상됩니다.

Git 충돌도 줄어드는데, 팀원 A는 users.py를, 팀원 B는 products.py를 동시에 수정해도 문제없습니다. 테스트도 쉬워지는데, 각 라우터를 독립적으로 테스트할 수 있어 단위 테스트가 간결해집니다.

또한 마이크로서비스로 전환할 때도 유리한데, 각 라우터를 별도의 서비스로 분리하기만 하면 됩니다. API 버전 관리도 가능해서, /api/v1/users와 /api/v2/users를 다른 라우터로 구현하여 하위 호환성을 유지할 수 있습니다.

실전 팁

💡 라우터마다 공통 의존성이 있으면 APIRouter(dependencies=[Depends(...)])로 모든 엔드포인트에 자동 적용할 수 있습니다.

💡 중첩 라우터도 가능합니다. 예: /api 라우터 안에 /users, /products 라우터를 넣어 /api/users/me 같은 구조를 만들 수 있습니다.

💡 라우터별로 다른 응답 모델을 설정할 수 있어서, 예를 들어 관리자 API는 더 많은 정보를, 일반 API는 제한된 정보만 반환하게 할 수 있습니다.

💡 프로젝트 구조는 app/routers/, app/models/, app/services/ 같은 폴더로 나누면 깔끔합니다. 도메인별로 폴더를 만들어도 좋습니다.

💡 Swagger 문서에서 tags_metadata를 사용하면 각 태그에 설명을 추가할 수 있어 API 문서 품질을 높일 수 있습니다.

---

8. 데이터베이스 연동

시작하며

여러분이 API에서 데이터를 저장하고 조회할 때 SQLAlchemy 같은 ORM을 어떻게 효율적으로 사용해야 할지 고민되지 않나요? 데이터베이스 세션을 언제 열고 닫아야 하는지, 트랜잭션은 어떻게 관리하는지, 여러 요청이 동시에 들어올 때 연결 풀은 어떻게 설정하는지 등 신경 써야 할 것이 많습니다.

이런 문제는 데이터베이스가 외부 리소스이기 때문에 발생합니다. 세션을 제대로 닫지 않으면 연결이 누수되어 결국 DB에 연결할 수 없게 되고, 트랜잭션을 잘못 관리하면 데이터 불일치가 발생합니다.

또한 비동기 처리와 함께 사용할 때는 더욱 복잡해집니다. 바로 이럴 때 필요한 것이 FastAPI와 SQLAlchemy의 올바른 통합 패턴입니다.

의존성 주입으로 세션을 관리하고, 컨텍스트 매니저로 자동 정리를 보장하며, 비동기 엔진을 사용해서 성능을 극대화할 수 있습니다. 이렇게 하면 안전하고 효율적인 데이터베이스 연동을 구현할 수 있습니다.

개요

간단히 말해서, 데이터베이스 연동은 ORM을 사용해서 데이터베이스 작업을 안전하고 효율적으로 수행하는 방법이며, FastAPI의 의존성 주입과 결합하면 세션 관리가 자동화됩니다. SQLAlchemy는 Python의 대표적인 ORM으로, SQL 쿼리를 Python 코드로 작성할 수 있게 해줍니다.

FastAPI와 함께 사용할 때는 의존성으로 세션을 주입하는 패턴이 일반적이며, 이렇게 하면 요청마다 새로운 세션을 생성하고, 요청 완료 후 자동으로 닫아줍니다. 예를 들어, get_db() 의존성이 세션을 yield하면 FastAPI가 알아서 정리합니다.

기존에는 각 엔드포인트에서 db = Session(), try-finally db.close() 같은 코드를 반복했다면, 이제는 의존성 한 번만 정의하면 모든 곳에서 재사용할 수 있습니다. 또한 트랜잭션도 자동으로 관리되어 에러 발생 시 롤백됩니다.

데이터베이스 연동의 핵심 특징은 세션 자동 관리, 트랜잭션 안전성, 그리고 ORM의 편리함입니다. 이러한 특징들은 복잡한 쿼리를 Python 코드로 작성할 수 있게 하고, SQL 인젝션 같은 보안 문제도 방지하며, 데이터베이스를 바꿀 때도 코드 수정이 최소화됩니다.

코드 예제

# database.py - 데이터베이스 설정
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker, Session
from sqlalchemy.ext.declarative import declarative_base

# 데이터베이스 연결 URL
DATABASE_URL = "postgresql://user:password@localhost/dbname"

# 엔진 생성 (연결 풀 포함)
engine = create_engine(DATABASE_URL, pool_pre_ping=True)

# 세션 팩토리
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 모델 베이스 클래스
Base = declarative_base()

# 의존성: DB 세션 제공
def get_db():
    db = SessionLocal()  # 세션 생성
    try:
        yield db  # 엔드포인트에 전달
    finally:
        db.close()  # 자동으로 닫기

# models.py - SQLAlchemy 모델
from sqlalchemy import Column, Integer, String
from database import Base

class User(Base):
    __tablename__ = "users"

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

# main.py - 엔드포인트에서 사용
from fastapi import FastAPI, Depends
from sqlalchemy.orm import Session
from database import get_db
from models import User

app = FastAPI()

@app.post("/users")
def create_user(email: str, name: str, db: Session = Depends(get_db)):
    # ORM으로 데이터 저장
    user = User(email=email, name=name)
    db.add(user)
    db.commit()  # 트랜잭션 커밋
    db.refresh(user)  # ID 등 DB에서 생성된 값 가져오기
    return user

@app.get("/users/{user_id}")
def get_user(user_id: int, db: Session = Depends(get_db)):
    # ORM으로 데이터 조회
    user = db.query(User).filter(User.id == user_id).first()
    if not user:
        raise HTTPException(status_code=404, detail="사용자 없음")
    return user

설명

이것이 하는 일: 데이터베이스 연동은 애플리케이션과 데이터베이스 사이의 연결을 관리하고, Python 객체로 데이터를 다루며, 트랜잭션을 자동으로 제어하여 데이터 일관성을 보장합니다. 첫 번째로, create_engine()으로 데이터베이스 엔진을 생성합니다.

이것은 실제 DB 연결을 관리하는 연결 풀을 포함하며, pool_pre_ping=True는 연결이 살아 있는지 체크해서 끊긴 연결을 자동으로 재연결합니다. sessionmaker()는 세션을 생성하는 팩토리 함수를 만들며, autocommit=False는 명시적으로 커밋해야 함을 의미합니다.

왜 이렇게 하냐면, 트랜잭션을 개발자가 제어하여 여러 작업을 하나의 원자적 단위로 묶을 수 있기 때문입니다. 그 다음으로, get_db() 의존성 함수가 핵심입니다.

이 함수는 세션을 생성해서 yield하고, 엔드포인트가 완료되면 finally 블록이 실행되어 세션을 닫습니다. 이것은 어떤 상황(정상 종료, 예외 발생)에서도 세션이 확실히 닫히게 보장합니다.

FastAPI의 의존성 주입 시스템은 이 패턴을 완벽하게 지원하며, 각 요청마다 독립적인 세션을 제공하므로 동시성 문제가 없습니다. 마지막으로, 엔드포인트에서는 db: Session = Depends(get_db)로 세션을 주입받아 사용합니다.

db.add(user)는 세션에 객체를 추가하고(아직 DB에 저장 안 됨), db.commit()이 실제로 INSERT 쿼리를 실행합니다. 만약 커밋 전에 예외가 발생하면 자동으로 롤백되어 데이터 일관성이 유지됩니다.

db.query(User).filter(...).first()는 SELECT 쿼리를 실행하며, SQLAlchemy가 자동으로 SQL을 생성하고 결과를 Python 객체로 변환합니다. 여러분이 이 코드를 사용하면 SQL을 직접 작성하지 않고도 데이터베이스를 안전하게 다룰 수 있으며, ORM이 자동으로 파라미터를 이스케이프해서 SQL 인젝션 공격을 방지합니다.

데이터베이스를 PostgreSQL에서 MySQL로 바꿔도 모델 코드는 거의 그대로 사용할 수 있어 이식성이 높습니다. 세션 관리도 자동화되어 연결 누수를 걱정할 필요가 없고, 트랜잭션도 안전하게 처리되어 데이터 불일치가 발생하지 않습니다.

또한 테스트 시 인메모리 SQLite를 사용하면 실제 DB 없이도 테스트할 수 있습니다. 비동기 처리가 필요하면 sqlalchemy.ext.asyncio를 사용하여 AsyncSession과 async def get_db()로 전환할 수 있습니다.

실전 팁

💡 pool_pre_ping=True는 필수입니다. DB 연결이 타임아웃으로 끊겨도 자동으로 재연결하여 에러를 방지합니다.

💡 비동기 FastAPI를 사용한다면 asyncpg 드라이버와 AsyncSession을 사용해서 성능을 크게 향상시킬 수 있습니다.

💡 대량의 데이터를 읽을 때는 db.query(...).yield_per(1000)로 청크 단위로 가져와서 메모리를 절약하세요.

💡 관계형 데이터(1:N, M:N)는 relationship()으로 정의하면 SQLAlchemy가 자동으로 조인해서 가져옵니다. 하지만 N+1 쿼리 문제를 조심하세요.

💡 Alembic을 사용해서 마이그레이션을 관리하면 스키마 변경을 버전 관리하고 자동으로 적용할 수 있어 배포가 안전해집니다. 이상으로 FastAPI 실무 활용 완벽 가이드를 마칩니다. 각 개념을 실무에 적용하여 더 안전하고 효율적인 API를 개발하시길 바랍니다!

---