Loading...
Loading...
02-reusable-code-python/api/fastapi_dependencies.py
"""
FastAPI 의존성 주입 패턴 - DB 세션, 인증, 페이지네이션, 설정 주입
@source GitHub-커뮤니티
@extracted 2026-02-16
@version 1.0.0
의존성:
- fastapi (필수, pip install fastapi)
- sqlalchemy (권장, pip install sqlalchemy[asyncio])
- pydantic (필수)
사용법:
from api.fastapi_dependencies import get_pagination, require_auth
@app.get("/users")
async def list_users(pagination: PaginationParams = Depends(get_pagination)):
return await user_service.list(pagination)
"""
import logging
from dataclasses import dataclass
from typing import Annotated, Any
from fastapi import Depends, Header, HTTPException, Query, Request, status
logger = logging.getLogger(__name__)
# ============================================
# 페이지네이션
# ============================================
@dataclass
class PaginationParams:
"""페이지네이션 파라미터"""
page: int
limit: int
offset: int
def get_pagination(
page: Annotated[int, Query(ge=1, description="페이지 번호")] = 1,
limit: Annotated[int, Query(ge=1, le=100, description="페이지당 항목 수")] = 10,
) -> PaginationParams:
"""페이지네이션 의존성
Example:
@app.get("/users")
async def list_users(
pagination: PaginationParams = Depends(get_pagination),
):
users = await db.find_many(
skip=pagination.offset,
take=pagination.limit,
)
"""
return PaginationParams(
page=page,
limit=limit,
offset=(page - 1) * limit,
)
# ============================================
# 정렬
# ============================================
@dataclass
class SortParams:
"""정렬 파라미터"""
sort_by: str
order: str
def get_sort(
sort_by: Annotated[str, Query(description="정렬 기준 필드")] = "created_at",
order: Annotated[str, Query(regex="^(asc|desc)$", description="정렬 방향")] = "desc",
) -> SortParams:
"""정렬 의존성
Example:
@app.get("/posts")
async def list_posts(sort: SortParams = Depends(get_sort)):
posts = await db.find_many(order_by={sort.sort_by: sort.order})
"""
return SortParams(sort_by=sort_by, order=order)
# ============================================
# 인증
# ============================================
@dataclass
class AuthUser:
"""인증된 사용자"""
id: str
email: str
role: str
# 세션/토큰 검증 함수 (프로젝트별로 교체)
async def _verify_token(token: str) -> AuthUser | None:
"""토큰 검증 (프로젝트별로 구현 교체 필요)"""
# 예시: JWT 검증 로직
# payload = jwt.decode(token, SECRET_KEY)
# return AuthUser(id=payload["sub"], email=payload["email"], role=payload["role"])
raise NotImplementedError("프로젝트별로 _verify_token 구현 필요")
async def get_current_user(
authorization: Annotated[str | None, Header()] = None,
) -> AuthUser:
"""현재 인증된 사용자 의존성
Example:
@app.get("/me")
async def get_me(user: AuthUser = Depends(get_current_user)):
return {"id": user.id, "email": user.email}
"""
if not authorization or not authorization.startswith("Bearer "):
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="인증이 필요합니다",
headers={"WWW-Authenticate": "Bearer"},
)
token = authorization.removeprefix("Bearer ")
user = await _verify_token(token)
if not user:
raise HTTPException(
status_code=status.HTTP_401_UNAUTHORIZED,
detail="유효하지 않은 토큰입니다",
)
return user
async def get_optional_user(
authorization: Annotated[str | None, Header()] = None,
) -> AuthUser | None:
"""선택적 인증 의존성 (비인증도 허용)
Example:
@app.get("/posts")
async def list_posts(user: AuthUser | None = Depends(get_optional_user)):
if user:
# 인증된 사용자 전용 로직
pass
"""
if not authorization or not authorization.startswith("Bearer "):
return None
try:
token = authorization.removeprefix("Bearer ")
return await _verify_token(token)
except Exception:
return None
def require_role(*roles: str):
"""역할 기반 권한 검사 의존성 팩토리
Example:
@app.delete("/users/{user_id}", dependencies=[Depends(require_role("admin"))])
async def delete_user(user_id: str):
await user_service.delete(user_id)
"""
async def _check_role(user: AuthUser = Depends(get_current_user)) -> AuthUser:
if user.role not in roles:
raise HTTPException(
status_code=status.HTTP_403_FORBIDDEN,
detail=f"권한이 부족합니다. 필요 역할: {', '.join(roles)}",
)
return user
return _check_role
# ============================================
# 설정 주입
# ============================================
def get_settings_dependency(settings_class: type):
"""설정 클래스 의존성 팩토리
Example:
from models.pydantic_patterns import AppSettings
get_settings = get_settings_dependency(AppSettings)
@app.get("/health")
async def health(settings = Depends(get_settings)):
return {"debug": settings.debug}
"""
_instance: Any = None
def _get():
nonlocal _instance
if _instance is None:
_instance = settings_class()
return _instance
return _get
# ============================================
# 레이트 리미팅 (간단 구현)
# ============================================
_rate_limit_store: dict[str, list[float]] = {}
def rate_limit(max_requests: int = 60, window_seconds: int = 60):
"""간단한 인메모리 레이트 리미팅
프로덕션에서는 Redis 기반으로 교체 권장
Example:
@app.post("/api/submit", dependencies=[Depends(rate_limit(10, 60))])
async def submit():
pass
"""
import time
async def _check(request: Request):
client_ip = request.client.host if request.client else "unknown"
now = time.time()
key = f"{client_ip}:{request.url.path}"
# 윈도우 내 요청 기록 필터
requests = _rate_limit_store.get(key, [])
requests = [t for t in requests if now - t < window_seconds]
if len(requests) >= max_requests:
raise HTTPException(
status_code=status.HTTP_429_TOO_MANY_REQUESTS,
detail="요청이 너무 많습니다. 잠시 후 다시 시도해주세요.",
)
requests.append(now)
_rate_limit_store[key] = requests
return _check