설정

최소 앱

from fastapi import FastAPI
app = FastAPI()

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

앱 실행

pip install "fastapi[standard]"
fastapi dev main.py    # dev with auto-reload
fastapi run main.py    # production

주요 기능

Async native	ASGI(Uvicorn)와 async/await
Auto docs	`/docs`에 Swagger UI, `/redoc`에 ReDoc
Type validation	요청/응답에 Pydantic 모델
OpenAPI	자동 생성 OpenAPI 스키마
Dependency injection	내장 DI 시스템

경로 작업

HTTP 메서드

@app.get("/items")
@app.post("/items")
@app.put("/items/{item_id}")
@app.patch("/items/{item_id}")
@app.delete("/items/{item_id}")

경로 파라미터

@app.get("/users/{user_id}")
async def get_user(user_id: int):
    return {"user_id": user_id}

# Enum constraint
from enum import Enum
class Color(str, Enum):
    red = "red"
    blue = "blue"

상태 코드 & 태그

from fastapi import status

@app.post("/items", status_code=status.HTTP_201_CREATED,
          tags=["items"])
async def create_item(item: Item):
    return item

요청 본문

Pydantic 모델

from pydantic import BaseModel, Field

class Item(BaseModel):
    name: str
    price: float = Field(gt=0, description="Must be positive")
    tags: list[str] = []

중첩 모델

class Address(BaseModel):
    street: str
    city: str
    zip_code: str

class User(BaseModel):
    name: str
    address: Address

엔드포인트에서 사용

@app.post("/items")
async def create_item(item: Item):
    return {"name": item.name, "price": item.price}

유효성 검사 기능

Field(gt=0)	0보다 큰 값
Field(min_length=1)	최소 문자열 길이
Field(max_length=100)	최대 문자열 길이
Field(pattern='^[a-z]+$')	정규식 패턴 일치
Field(default=None)	기본값이 있는 선택적 필드
EmailStr	이메일 유효성 검사 (pydantic[email])

쿼리 파라미터

기본 쿼리 파라미터

@app.get("/items")
async def list_items(skip: int = 0, limit: int = 10):
    return items[skip : skip + limit]
# GET /items?skip=0&limit=20

쿼리 유효성 검사

from fastapi import Query

@app.get("/search")
async def search(
    q: str = Query(min_length=3, max_length=50),
    page: int = Query(default=1, ge=1),
):
    return {"q": q, "page": page}

선택적 & 필수

async def read_items(
    q: str | None = None,   # optional
    name: str = ...,         # required (Ellipsis)
    tags: list[str] = Query(default=[]),
):
    return {"q": q, "name": name}

헤더 & 쿠키

from fastapi import Header, Cookie

async def read(
    user_agent: str | None = Header(default=None),
    session_id: str | None = Cookie(default=None),
):
    return {"ua": user_agent}

응답 모델

class ItemOut(BaseModel):
    name: str
    price: float

@app.get("/items/{id}", response_model=ItemOut)
async def get_item(id: int):
    return items[id]  # filters out extra fields

복수 응답 타입

from fastapi.responses import JSONResponse, HTMLResponse

@app.get("/html", response_class=HTMLResponse)
async def get_html():
    return "Hello"

응답 모델 옵션

response_model	출력 필터링용 Pydantic 모델
response_model_exclude_unset	명시적으로 설정하지 않은 필드 제외
response_model_include	특정 필드만 포함 (화이트리스트)
response_model_exclude	특정 필드 제외 (블랙리스트)

에러 응답

from fastapi import HTTPException

@app.get("/items/{id}")
async def get_item(id: int):
    if id not in items:
        raise HTTPException(status_code=404, detail="Not found")
    return items[id]

의존성

함수 의존성

from fastapi import Depends

async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

엔드포인트에서 사용

@app.get("/users")
async def list_users(db: Session = Depends(get_db)):
    return db.query(User).all()

클래스 기반 의존성

class Pagination:
    def __init__(self, skip: int = 0, limit: int = 10):
        self.skip = skip
        self.limit = limit

@app.get("/items")
async def list_items(pg: Pagination = Depends()):
    return items[pg.skip : pg.skip + pg.limit]

의존성 범위

Depends(func)	엔드포인트별 의존성
app = FastAPI(dependencies=[...])	모든 라우트의 전역 의존성
APIRouter(dependencies=[...])	라우터 수준 의존성
yield	설정/해제 (DB 세션, 잠금)

인증

OAuth2 Password Bearer

from fastapi.security import OAuth2PasswordBearer

oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

@app.get("/users/me")
async def read_me(token: str = Depends(oauth2_scheme)):
    user = decode_token(token)
    return user

JWT 토큰 흐름

from jose import jwt
SECRET = "your-secret-key"

def create_token(data: dict):
    return jwt.encode(data, SECRET, algorithm="HS256")

def decode_token(token: str):
    return jwt.decode(token, SECRET, algorithms=["HS256"])

토큰 엔드포인트

from fastapi.security import OAuth2PasswordRequestForm

@app.post("/token")
async def login(form: OAuth2PasswordRequestForm = Depends()):
    user = authenticate(form.username, form.password)
    if not user:
        raise HTTPException(status_code=401)
    return {"access_token": create_token({"sub": user.id})}

보안 스킴

OAuth2PasswordBearer	폼 로그인을 통한 Bearer 토큰
HTTPBasic	기본 사용자명/비밀번호 인증
APIKeyHeader	헤더의 API 키
APIKeyCookie	쿠키의 API 키

백그라운드 작업

간단한 백그라운드 작업

from fastapi import BackgroundTasks def send_email(to: str, body: str): # slow operation runs after response email_client.send(to, body) @app.post("/notify") async def notify(bg: BackgroundTasks): bg.add_task(send_email, "[email protected]", "Hello!") return {"status": "queued"}

백그라운드를 포함한 의존성

async def log_request(bg: BackgroundTasks):
    bg.add_task(write_log, "request received")

@app.get("/items", dependencies=[Depends(log_request)])
async def list_items():
    return items

백그라운드 vs 워커

BackgroundTasks	응답 후 가벼운 작업 (이메일, 로그)
Celery / ARQ	별도 워커가 필요한 무거운 작업
asyncio.create_task	fire-and-forget 비동기 코루틴

미들웨어

커스텀 미들웨어

import time
from starlette.middleware.base import BaseHTTPMiddleware

class TimingMiddleware(BaseHTTPMiddleware):
    async def dispatch(self, request, call_next):
        start = time.time()
        response = await call_next(request)
        duration = time.time() - start
        response.headers["X-Process-Time"] = str(duration)
        return response

미들웨어 추가

app.add_middleware(TimingMiddleware)

CORS

from fastapi.middleware.cors import CORSMiddleware

app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_methods=["*"],
    allow_headers=["*"],
)

내장 미들웨어

CORSMiddleware	크로스 오리진 리소스 공유
TrustedHostMiddleware	허용된 호스트명 제한
GZipMiddleware	Gzip 응답 압축
HTTPSRedirectMiddleware	HTTP를 HTTPS로 리다이렉트

테스트

테스트 클라이언트

from fastapi.testclient import TestClient

client = TestClient(app)

def test_read_root():
    resp = client.get("/")
    assert resp.status_code == 200
    assert resp.json() == {"message": "Hello, World!"}

POST 테스트

def test_create_item():
    resp = client.post("/items", json={
        "name": "Widget",
        "price": 9.99,
    })
    assert resp.status_code == 201
    assert resp.json()["name"] == "Widget"

의존성 재정의

async def mock_db():
    return FakeDB()

app.dependency_overrides[get_db] = mock_db

def test_with_mock_db():
    resp = client.get("/users")
    assert resp.status_code == 200

비동기 테스트

import pytest
from httpx import AsyncClient, ASGITransport

@pytest.mark.anyio
async def test_async():
    transport = ASGITransport(app=app)
    async with AsyncClient(transport=transport) as ac:
        resp = await ac.get("/")
    assert resp.status_code == 200

FASTAPI 빠른 참조

Hello