Python FastAPI 개념 정리

1️⃣ FastAPI 기초

FastAPI는 현대적이고 빠른 웹 프레임워크이다.

from fastapi import FastAPI, HTTPException, Query, Path
from pydantic import BaseModel, Field, EmailStr
from typing import List, Optional

app = FastAPI(
    title="My API",
    description="API 설명",
    version="0.1.0"
)

class User(BaseModel):
    id: int
    name: str
    email: EmailStr
    is_active: bool = True
    
    class Config:
        schema_extra = {
            "example": {
                "id": 1,
                "name": "홍길동",
                "email": "[email protected]",
                "is_active": True
            }
        }

# 인메모리 데이터베이스
users_db = {}

# 기본 라우트
@app.get("/")
async def root():
    return {"message": "Hello World"}

# 경로 매개변수
@app.get("/users/{user_id}", response_model=User)
async def get_user(user_id: int = Path(..., title="사용자 ID", ge=1)):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")
    return users_db[user_id]

# 요청 본문
@app.post("/users/", response_model=User, status_code=201)
async def create_user(user: User):
    if user.id in users_db:
        raise HTTPException(status_code=400, detail="이미 존재하는 사용자 ID입니다")
    users_db[user.id] = user
    return user

# 쿼리 매개변수
@app.get("/users/", response_model=List[User])
async def list_users(
    skip: int = Query(0, ge=0),
    limit: int = Query(10, ge=1, le=100)
):
    return list(users_db.values())[skip : skip + limit]

# PUT 요청
@app.put("/users/{user_id}", response_model=User)
async def update_user(user_id: int, user: User):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")
    users_db[user_id] = user
    return user

# DELETE 요청
@app.delete("/users/{user_id}", status_code=204)
async def delete_user(user_id: int):
    if user_id not in users_db:
        raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")
    del users_db[user_id]
    return None

✅ 특징:

빠른 성능
자동 API 문서화
타입 힌트 지원
OpenAPI 및 JSON 스키마 자동 생성
Python 3.6+ 비동기 기능 활용

문서화 자동 생성:

FastAPI는 자동으로 /docs와 /redoc 엔드포인트를 제공하여 API 문서를 생성한다.

/docs: Swagger UI를 이용한 대화형 API 문서
/redoc: ReDoc을 이용한 문서

2️⃣ 의존성 주입

FastAPI의 강력한 의존성 주입 시스템을 활용한 코드 구성 방법이다.

from fastapi import Depends, HTTPException, FastAPI
from sqlalchemy.orm import Session
from typing import Annotated, List, Optional
from database import SessionLocal, engine, Base
from models import UserModel
from schemas import UserCreate, UserResponse

app = FastAPI()

# 데이터베이스 의존성
async def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

DB = Annotated[Session, Depends(get_db)]

# 서비스 클래스
class UserService:
    def __init__(self, db: Session = Depends(get_db)):
        self.db = db
    
    def get_user(self, user_id: int):
        user = self.db.query(UserModel).filter(
            UserModel.id == user_id
        ).first()
        if not user:
            raise HTTPException(
                status_code=404, 
                detail="사용자를 찾을 수 없습니다"
            )
        return user
    
    def create_user(self, user: UserCreate):
        db_user = UserModel(**user.dict())
        self.db.add(db_user)
        self.db.commit()
        self.db.refresh(db_user)
        return db_user
    
    def get_users(self, skip: int = 0, limit: int = 100):
        return self.db.query(UserModel).offset(skip).limit(limit).all()

# 컨트롤러에서 의존성 주입 활용
@app.get("/users/{user_id}", response_model=UserResponse)
async def get_user(
    user_id: int,
    user_service: UserService = Depends()
):
    return user_service.get_user(user_id)

@app.post("/users/", response_model=UserResponse)
async def create_user(
    user: UserCreate,
    user_service: UserService = Depends()
):
    return user_service.create_user(user)

# 함수 형태의 의존성
async def common_parameters(q: Optional[str] = None, skip: int = 0, limit: int = 100):
    return {"q": q, "skip": skip, "limit": limit}

@app.get("/items/")
async def read_items(commons: dict = Depends(common_parameters)):
    return {"params": commons}

# 의존성 체이닝
async def verify_token(x_token: str = Header(...)):
    if x_token != "fake-super-secret-token":
        raise HTTPException(status_code=400, detail="X-Token 헤더가 유효하지 않습니다")
    return x_token

async def verify_key(x_key: str = Header(...), token: str = Depends(verify_token)):
    if x_key != "fake-super-secret-key":
        raise HTTPException(status_code=400, detail="X-Key 헤더가 유효하지 않습니다")
    return x_key

@app.get("/secure-items/", dependencies=[Depends(verify_token), Depends(verify_key)])
async def read_secure_items():
    return [{"item": "Secure Item"}]

✅ 특징:

의존성 관리
데이터베이스 세션
서비스 계층 분리
재사용 가능한 의존성
의존성 체이닝

3️⃣ 미들웨어와 보안

애플리케이션 전반의 기능을 추가하는 미들웨어와 보안 설정이다.

from fastapi import FastAPI, Depends, HTTPException, Security, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from fastapi.middleware.cors import CORSMiddleware
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
import time
from typing import Dict

app = FastAPI()

# CORS 미들웨어
app.add_middleware(
    CORSMiddleware,
    allow_origins=["*"],  # 프로덕션에서는 특정 도메인 지정
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# 성능 측정 미들웨어
@app.middleware("http")
async def add_process_time_header(request, call_next):
    start_time = time.time()
    response = await call_next(request)
    process_time = time.time() - start_time
    response.headers["X-Process-Time"] = str(process_time)
    return response

# 보안 설정
pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

SECRET_KEY = "YOUR_SECRET_KEY_HERE"  # 프로덕션에서는 환경변수 사용
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

def verify_password(plain_password, hashed_password):
    return pwd_context.verify(plain_password, hashed_password)

def get_password_hash(password):
    return pwd_context.hash(password)

def create_access_token(data: dict, expires_delta: timedelta = None):
    to_encode = data.copy()
    if expires_delta:
        expire = datetime.utcnow() + expires_delta
    else:
        expire = datetime.utcnow() + timedelta(minutes=15)
    to_encode.update({"exp": expire})
    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
    return encoded_jwt

# 인증
fake_users_db = {
    "johndoe": {
        "username": "johndoe",
        "full_name": "John Doe",
        "email": "[email protected]",
        "hashed_password": get_password_hash("secret"),
        "disabled": False,
    }
}

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="인증할 수 없습니다",
        headers={"WWW-Authenticate": "Bearer"},
    )
    try:
        payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
        username: str = payload.get("sub")
        if username is None:
            raise credentials_exception
    except JWTError:
        raise credentials_exception
    
    user = fake_users_db.get(username)
    if user is None:
        raise credentials_exception
    return user

async def get_current_active_user(current_user: dict = Security(get_current_user)):
    if current_user.get("disabled"):
        raise HTTPException(status_code=400, detail="비활성화된 사용자입니다")
    return current_user

# 로그인 엔드포인트
@app.post("/token")
async def login_for_access_token(form_data: OAuth2PasswordRequestForm = Depends()):
    user = fake_users_db.get(form_data.username)
    if not user or not verify_password(form_data.password, user["hashed_password"]):
        raise HTTPException(
            status_code=status.HTTP_401_UNAUTHORIZED,
            detail="유효하지 않은 사용자명 또는 비밀번호입니다",
            headers={"WWW-Authenticate": "Bearer"},
        )
    
    access_token_expires = timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    access_token = create_access_token(
        data={"sub": user["username"]}, expires_delta=access_token_expires
    )
    return {"access_token": access_token, "token_type": "bearer"}

# 보호된 엔드포인트
@app.get("/users/me", response_model=Dict)
async def read_users_me(current_user: dict = Depends(get_current_active_user)):
    return current_user

✅ 특징:

JWT 인증
미들웨어 처리
보안 설정
암호화
CORS 설정

4️⃣ 비동기 처리

FastAPI의 비동기 기능을 활용한 고성능 애플리케이션 구현이다.

import asyncio
import httpx
from fastapi import FastAPI, BackgroundTasks, Depends
from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.future import select
from typing import List

app = FastAPI()

# 비동기 데이터베이스 설정
DATABASE_URL = "postgresql+asyncpg://user:password@localhost/dbname"
async_engine = create_async_engine(DATABASE_URL)
AsyncSessionLocal = sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

async def get_async_session():
    async with AsyncSessionLocal() as session:
        yield session

# 백그라운드 작업
def write_log(message: str):
    with open("log.txt", mode="a") as log:
        log.write(f"{message}\n")

async def send_email_async(email: str, message: str):
    # 이메일 전송 시뮬레이션
    await asyncio.sleep(1)  # API 호출 시뮬레이션
    return {"email": email, "message": message, "status": "sent"}

@app.post("/send-notification/{email}")
async def send_notification(
    email: str, 
    background_tasks: BackgroundTasks
):
    # 백그라운드에서 로그 기록
    background_tasks.add_task(write_log, f"알림 전송: {email}")
    # 백그라운드에서 이메일 전송
    background_tasks.add_task(send_email_async, email, "중요 알림입니다")
    return {"message": "알림이 전송되었습니다"}

# 비동기 HTTP 요청
@app.get("/github-stars/{username}")
async def get_github_stars(username: str):
    async with httpx.AsyncClient() as client:
        response = await client.get(f"https://api.github.com/users/{username}")
        if response.status_code != 200:
            return {"error": "GitHub API 호출 실패"}
        
        user_data = response.json()
        # 병렬 API 호출
        repos_response = await client.get(f"https://api.github.com/users/{username}/repos")
        repos = repos_response.json()
        
        total_stars = sum(repo["stargazers_count"] for repo in repos)
        return {
            "username": username,
            "repos_count": len(repos),
            "total_stars": total_stars
        }

# 비동기 데이터베이스 작업
async def get_user_async(async_session: AsyncSession, user_id: int):
    result = await async_session.execute(
        select(User).where(User.id == user_id)
    )
    return result.scalar_one_or_none()

@app.get("/users/{user_id}")
async def get_user(
    user_id: int,
    async_session: AsyncSession = Depends(get_async_session)
):
    user = await get_user_async(async_session, user_id)
    if not user:
        raise HTTPException(status_code=404, detail="사용자를 찾을 수 없습니다")
    return user

# 병렬 처리
@app.get("/parallel-tasks")
async def parallel_tasks():
    start_time = time.time()
    
    # 병렬로 비동기 작업 실행
    task1 = asyncio.create_task(asyncio.sleep(1))
    task2 = asyncio.create_task(asyncio.sleep(1))
    task3 = asyncio.create_task(asyncio.sleep(1))
    
    # 모든 작업 완료 대기
    await asyncio.gather(task1, task2, task3)
    
    end_time = time.time()
    return {
        "message": "모든 병렬 작업 완료",
        "time_taken": end_time - start_time  # 약 1초
    }

✅ 특징:

비동기 처리
백그라운드 작업
성능 최적화
병렬 API 호출
비동기 데이터베이스 연산

5️⃣ WebSocket과 이벤트 스트리밍

WebSocket 및 서버-전송 이벤트를 활용한 실시간 애플리케이션 구현이다.

from fastapi import FastAPI, WebSocket, WebSocketDisconnect, Depends
from fastapi.responses import StreamingResponse
import asyncio
import json
from datetime import datetime
from typing import List, Dict

app = FastAPI()

# WebSocket 관리자
class ConnectionManager:
    def __init__(self):
        self.active_connections: List[WebSocket] = []
    
    async def connect(self, websocket: WebSocket):
        await websocket.accept()
        self.active_connections.append(websocket)
    
    def disconnect(self, websocket: WebSocket):
        self.active_connections.remove(websocket)
    
    async def send_personal_message(self, message: str, websocket: WebSocket):
        await websocket.send_text(message)
    
    async def broadcast(self, message: str):
        for connection in self.active_connections:
            await connection.send_text(message)

manager = ConnectionManager()

# WebSocket 엔드포인트
@app.websocket("/ws/{client_id}")
async def websocket_endpoint(websocket: WebSocket, client_id: int):
    await manager.connect(websocket)
    try:
        while True:
            data = await websocket.receive_text()
            # 개인 메시지 전송
            await manager.send_personal_message(f"당신: {data}", websocket)
            # 브로드캐스트 메시지 전송
            await manager.broadcast(f"클라이언트 #{client_id}의 메시지: {data}")
    except WebSocketDisconnect:
        manager.disconnect(websocket)
        await manager.broadcast(f"클라이언트 #{client_id}가 퇴장했습니다")

# 서버-전송 이벤트 (SSE)
@app.get("/sse")
async def sse():
    async def event_generator():
        for i in range(10):
            # JSON 데이터 생성
            data = json.dumps({
                "id": i,
                "timestamp": datetime.now().isoformat(),
                "message": f"이벤트 #{i}"
            })
            yield f"data: {data}\n\n"
            await asyncio.sleep(1)
    
    return StreamingResponse(
        event_generator(),
        media_type="text/event-stream"
    )

# WebSocket 클라이언트 예제 (JavaScript)
"""
const ws = new WebSocket("ws://localhost:8000/ws/1");
ws.onmessage = function(event) {
    console.log('메시지 수신:', event.data);
};
ws.onclose = function(event) {
    console.log('연결 종료:', event);
};
ws.send("안녕하세요!");

// SSE 클라이언트
const eventSource = new EventSource("/sse");
eventSource.onmessage = (event) => {
    const data = JSON.parse(event.data);
    console.log("이벤트 수신:", data);
};
eventSource.onerror = (error) => {
    console.error("SSE 오류:", error);
    eventSource.close();
};
"""

✅ 특징:

양방향 실시간 통신
서버-전송 이벤트
메시지 브로드캐스팅
연결 관리
비동기 이벤트 스트리밍

6️⃣ 테스팅과 배포

FastAPI 애플리케이션의 테스트 및 배포 방법이다.

# test_main.py
from fastapi.testclient import TestClient
from main import app

client = TestClient(app)

def test_read_main():
    response = client.get("/")
    assert response.status_code == 200
    assert response.json() == {"message": "Hello World"}

def test_create_user():
    user_data = {
        "id": 1,
        "name": "Test User",
        "email": "[email protected]",
        "is_active": True
    }
    response = client.post("/users/", json=user_data)
    assert response.status_code == 201
    assert response.json()["id"] == user_data["id"]
    assert response.json()["name"] == user_data["name"]

def test_get_user():
    # 먼저 사용자 생성
    user_data = {
        "id": 2,
        "name": "Another User",
        "email": "[email protected]",
        "is_active": True
    }
    client.post("/users/", json=user_data)
    
    # 사용자 조회
    response = client.get("/users/2")
    assert response.status_code == 200
    assert response.json()["name"] == user_data["name"]

def test_get_nonexistent_user():
    response = client.get("/users/999")
    assert response.status_code == 404

Dockerfile:

FROM python:3.9-slim

WORKDIR /app

COPY requirements.txt .
RUN pip install --no-cache-dir -r requirements.txt

COPY . .

# production 설정
ENV PORT=8000
ENV HOST=0.0.0.0
ENV WORKERS=4

CMD ["uvicorn", "main:app", "--host", "0.0.0.0", "--port", "8000", "--workers", "4"]

배포 구성:

# gunicorn_conf.py
from multiprocessing import cpu_count

# Gunicorn 설정
bind = "0.0.0.0:8000"
workers = cpu_count() * 2 + 1
worker_class = "uvicorn.workers.UvicornWorker"
keepalive = 120
timeout = 120
graceful_timeout = 120
max_requests = 10000
max_requests_jitter = 1000

# 배포 명령어
gunicorn -c gunicorn_conf.py main:app

✅ 특징:

자동화된 테스트
컨테이너화
성능 최적화
프로덕션 설정
CI/CD 통합

주요 팁

✅ 모범 사례:

타입 힌트 활용: Pydantic 모델 및 타입 어노테이션으로 코드 안정성 향상
의존성 주입 활용: 재사용 가능한 컴포넌트 개발 및 테스트 용이성 증대
비동기 처리 최적화: 동시성 및 I/O 바운드 작업 최적화로 성능 향상
문서화 자동화: OpenAPI 스키마 및 Swagger UI로 API 문서 자동화
테스트 코드 작성: TestClient를 활용한 엔드포인트 테스트 자동화
에러 처리 표준화: 일관된 에러 응답으로 클라이언트 개발 편의성 향상
캐싱 전략 수립: Redis나 FastAPI 캐싱 미들웨어로 성능 최적화
보안 설정 확인: 적절한 인증, 권한 확인 및 CORS 설정 구현
환경 변수 관리: 개발, 테스트, 프로덕션 환경별 설정 분리
로깅 시스템 구축: 구조화된 로깅으로 디버깅 및 모니터링 강화
비즈니스 로직 분리: 컨트롤러와 서비스 레이어 분리로 코드 조직화
마이그레이션 관리: Alembic을 활용한 데이터베이스 스키마 관리
API 버전 관리: 경로나 헤더 기반 API 버전 관리 전략 수립
레이트 리미팅: 트래픽 제한으로 서비스 안정성 보장
헬스 체크 구현: 애플리케이션 상태 모니터링 엔드포인트 제공

KR_FastAPI - somaz94/python-study GitHub Wiki

Python FastAPI 개념 정리

1️⃣ FastAPI 기초

문서화 자동 생성:

2️⃣ 의존성 주입

3️⃣ 미들웨어와 보안

4️⃣ 비동기 처리

5️⃣ WebSocket과 이벤트 스트리밍

6️⃣ 테스팅과 배포

Dockerfile:

배포 구성:

주요 팁

⚠️ GitHub.com Fallback ⚠️

KR_FastAPI - somaz94/python-study GitHub Wiki

Python FastAPI 개념 정리

1️⃣ FastAPI 기초

문서화 자동 생성:

2️⃣ 의존성 주입

3️⃣ 미들웨어와 보안

4️⃣ 비동기 처리

5️⃣ WebSocket과 이벤트 스트리밍

6️⃣ 테스팅과 배포

Dockerfile:

배포 구성:

주요 팁

⚠️ **GitHub.com Fallback** ⚠️

⚠️ GitHub.com Fallback ⚠️