시작하기 전에: 실무에서 자주 마주치는 3가지 오류
AI API를 실무에 통합할 때, 대부분의 개발자가 처음 겪는 장애 패턴은 정해져 있습니다.
- ConnectionError: timeout — 동시에 다수의 API 요청이 발생하여 연결이 타임아웃
- 401 Unauthorized — API 키 만료 또는 잘못된 엔드포인트 설정으로 인증 실패
- 429 Too Many Requests — Rate Limit 초과로 인한 요청 거부
이 세 가지 문제는 사실 미들웨어 레이어에서 한 번에 해결할 수 있습니다. 이번 튜토리얼에서는 Python FastAPI 기반의 게이트웨이 미들웨어를 직접 구현하며, 인증·요청제한·로깅을 하나의 파이프라인으로 통합하는 방법을 다루겠습니다.
왜 일체화 미들웨어가 필요한가?
개별적으로 구현하면 각 모듈이 중복 체크를 하고, 상태 관리도 복잡해집니다. 하지만 하나의 미들웨어 스택으로 통합하면:
- 요청이 게이트웨이 진입 시 한 번만 인증 검증 → 불필요한 API 키 노출 방지
- 트래픽 패턴에 따라 동적으로 Rate Limit 조정 → 429 에러 최소화
- 모든 요청/응답을 구조화된 로깅 → 비용 추적과 장애 분석 용이
지금 가입해서 HolySheep AI의 글로벌 AI API 게이트웨이에서 이러한 미들웨어 패턴을 직접 체험해보세요. HolySheep AI는 단일 API 키로 GPT-4.1, Claude Sonnet 4, Gemini 2.5 Flash, DeepSeek V3.2 등 모든 주요 모델에 접근할 수 있습니다.
프로젝트 구조
ai_gateway/
├── main.py # FastAPI 앱 진입점
├── middleware/
│ ├── __init__.py
│ ├── auth.py # 인증 미들웨어
│ ├── rate_limiter.py # Rate Limit 미들웨어
│ └── logging_middleware.py # 로깅 미들웨어
├── models/
│ ├── __init__.py
│ └── request_models.py # 요청/응답 스키마
├── services/
│ ├── __init__.py
│ └── ai_client.py # HolySheep AI 연동 클라이언트
├── config.py # 설정 관리
└── requirements.txt
핵심 구현: 일체화 미들웨어 스택
1단계: 설정 및 의존성
# requirements.txt
fastapi==0.115.0
uvicorn==0.30.0
httpx==0.27.0
redis==5.0.0
python-jose[cryptography]==3.3.0
pydantic==2.6.0
pydantic-settings==2.2.0
# config.py
from pydantic_settings import BaseSettings
from typing import Optional
class Settings(BaseSettings):
# HolySheep AI 설정
HOLYSHEEP_BASE_URL: str = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY: str = "YOUR_HOLYSHEEP_API_KEY"
# Rate Limit 설정 (Redis 기반)
REDIS_HOST: str = "localhost"
REDIS_PORT: int = 6379
RATE_LIMIT_REQUESTS: int = 60 # 분당 요청 수
RATE_LIMIT_WINDOW: int = 60 # 윈도우 크기(초)
# 로깅 설정
LOG_LEVEL: str = "INFO"
LOG_FILE: str = "gateway.log"
class Config:
env_file = ".env"
settings = Settings()
2단계: 인증 미들웨어
# middleware/auth.py
from fastapi import Request, HTTPException, status
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from jose import jwt, JWTError
from typing import Callable
import logging
logger = logging.getLogger(__name__)
class AuthenticationMiddleware(BaseHTTPMiddleware):
"""
JWT 토큰 및 API 키 인증을 처리하는 미들웨어
인증 실패 시 발생하는 주요 오류:
- 401 Unauthorized: 토큰이 없거나 만료된 경우
- 403 Forbidden: 유효하지 않은 토큰 시그니처
"""
def __init__(self, app, api_key: str):
super().__init__(app)
self.api_key = api_key
self.allowed_paths = {"/health", "/docs", "/openapi.json"}
async def dispatch(self, request: Request, call_next: Callable):
# 건너뛸 경로 체크
if request.url.path in self.allowed_paths:
return await call_next(request)
# API 키 인증 (X-API-Key 헤더 또는 Authorization Bearer 토큰)
api_key = request.headers.get("X-API-Key")
auth_header = request.headers.get("Authorization", "")
if auth_header.startswith("Bearer "):
token = auth_header[7:]
try:
# HolySheep AI API 키를 JWT 형태로 검증
payload = jwt.decode(
token,
self.api_key,
algorithms=["HS256"]
)
request.state.user_id = payload.get("sub")
request.state.api_tier = payload.get("tier", "free")
except JWTError as e:
logger.error(f"JWT 검증 실패: {str(e)}")
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": "Invalid authentication token"}
)
elif api_key:
# 단순 API 키 인증
if api_key != self.api_key:
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": "Invalid API key"}
)
else:
return JSONResponse(
status_code=status.HTTP_401_UNAUTHORIZED,
content={"error": "Missing authentication credentials"}
)
return await call_next(request)
3단계: Rate Limit 미들웨어
# middleware/rate_limiter.py
from fastapi import Request
from fastapi.responses import JSONResponse
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Callable
import redis
import time
import logging
from config import settings
logger = logging.getLogger(__name__)
class RateLimitMiddleware(BaseHTTPMiddleware):
"""
Redis 기반 슬라이딩 윈도우 Rate Limiting 미들웨어
Rate Limit 초과 시 발생하는 오류:
- 429 Too Many Requests: 분당 요청 할당량 초과
- 503 Service Unavailable: Redis 연결 실패 시
"""
def __init__(self, app, requests_per_minute: int = 60):
super().__init__(app)
self.requests_per_minute = requests_per_minute
self.window_size = 60 # 1분 윈도우
try:
self.redis_client = redis.Redis(
host=settings.REDIS_HOST,
port=settings.REDIS_PORT,
decode_responses=True
)
# 연결 테스트
self.redis_client.ping()
except redis.ConnectionError as e:
logger.warning(f"Redis 연결 실패, 메모리 기반 폴백: {e}")
self.redis_client = None
self.local_cache = {}
async def dispatch(self, request: Request, call_next: Callable):
client_id = self._get_client_identifier(request)
current_time = int(time.time())
if not self._check_rate_limit(client_id, current_time):
logger.warning(f"Rate Limit 초과: {client_id}")
return JSONResponse(
status_code=429,
content={
"error": "Too Many Requests",
"message": f"Rate limit exceeded. Max {self.requests_per_minute} requests per minute.",
"retry_after": self.window_size
},
headers={"Retry-After": str(self.window_size)}
)
response = await call_next(request)
# 응답 헤더에 남은 할당량 정보 추가
remaining = self._get_remaining_requests(client_id, current_time)
response.headers["X-RateLimit-Remaining"] = str(remaining)
response.headers["X-RateLimit-Limit"] = str(self.requests_per_minute)
return response
def _get_client_identifier(self, request: Request) -> str:
"""클라이언트 식별자 추출 (API 키 기반)"""
return getattr(request.state, "user_id", "anonymous")
def _check_rate_limit(self, client_id: str, current_time: int) -> bool:
"""슬라이딩 윈도우 알고리즘으로 Rate Limit 체크"""
key = f"rate_limit:{client_id}"
window_start = current_time - self.window_size
if self.redis_client:
# Redis 기반 카운팅
pipe = self.redis_client.pipeline()
pipe.zremrangebyscore(key, 0, window_start)
pipe.zcard(key)
pipe.zadd(key, {str(current_time): current_time})
pipe.expire(key, self.window_size * 2)
results = pipe.execute()
request_count = results[1]
else:
# 메모리 폴백 (개발/테스트용)
if client_id not in self.local_cache:
self.local_cache[client_id] = []
# 오래된 요청 기록 제거
self.local_cache[client_id] = [
t for t in self.local_cache[client_id]
if t > window_start
]
request_count = len(self.local_cache[client_id])
self.local_cache[client_id].append(current_time)
return request_count < self.requests_per_minute
def _get_remaining_requests(self, client_id: str, current_time: int) -> int:
"""남은 요청 수 계산"""
key = f"rate_limit:{client_id}"
window_start = current_time - self.window_size
if self.redis_client:
self.redis_client.zremrangebyscore(key, 0, window_start)
count = self.redis_client.zcard(key)
else:
if client_id in self.local_cache:
count = len([t for t in self.local_cache[client_id] if t > window_start])
else:
count = 0
return max(0, self.requests_per_minute - count)
4단계: 로깅 미들웨어
# middleware/logging_middleware.py
from fastapi import Request
from starlette.middleware.base import BaseHTTPMiddleware
from typing import Callable
import time
import json
import logging
from datetime import datetime
from config import settings
logger = logging.getLogger(__name__)
class LoggingMiddleware(BaseHTTPMiddleware):
"""
구조화된 요청/응답 로깅 미들웨어
로깅 데이터로 분석 가능한 항목:
- 토큰 사용량 및 비용 추적
- 응답 시간 분포
- 에러 발생 패턴
- 모델별 요청 분포
"""
def __init__(self, app):
super().__init__(app)
# 파일 핸들러 설정
self.file_handler = logging.FileHandler(settings.LOG_FILE)
self.file_handler.setFormatter(
logging.Formatter('%(message)s')
)
logger.addHandler(self.file_handler)
logger.setLevel(settings.LOG_LEVEL)
async def dispatch(self, request: Request, call_next: Callable):
start_time = time.time()
request_id = f"{datetime.utcnow().timestamp()}-{id(request)}"
# 요청 정보 수집
request_data = {
"request_id": request_id,
"method": request.method,
"path": str(request.url.path),
"client": request.client.host if request.client else "unknown",
"user_id": getattr(request.state, "user_id", "anonymous"),
"timestamp": datetime.utcnow().isoformat(),
}
# 요청 본문 로깅 (바이너리 제외)
if request.method in ["POST", "PUT", "PATCH"]:
content_type = request.headers.get("content-type", "")
if "application/json" in content_type:
try:
body = await request.body()
if body:
parsed = json.loads(body)
# 민감 정보 마스킹
if "prompt" in parsed:
request_data["prompt_length"] = len(parsed["prompt"])
if "messages" in parsed:
request_data["message_count"] = len(parsed["messages"])
except Exception:
pass
# 요청 처리
response = await call_next(request)
process_time = (time.time() - start_time) * 1000 # ms 변환
# 응답 정보 추가
response_data = {
"request_id": request_id,
"status_code": response.status_code,
"process_time_ms": round(process_time, 2),
"model": response.headers.get("X-Used-Model", "unknown"),
"tokens_used": int(response.headers.get("X-Tokens-Used", 0)),
}
# 비용 계산 (HolySheep AI 공식 요금)
model_prices = {
"gpt-4.1": {"input": 8.0, "output": 32.0}, # $/MTok
"claude-sonnet-4": {"input": 15.0, "output": 75.0},
"gemini-2.5-flash": {"input": 2.50, "output": 10.0},
"deepseek-v3.2": {"input": 0.42, "output": 1.68},
}
model = response_data["model"].lower()
if model in model_prices and response_data["tokens_used"] > 0:
# 대략적인 비용 계산 (입력:출력 = 3:1 비율 가정)
input_tokens = response_data["tokens_used"] * 0.75
output_tokens = response_data["tokens_used"] * 0.25
price = model_prices[model]
response_data["estimated_cost_usd"] = round(
(input_tokens * price["input"] + output_tokens * price["output"]) / 1_000_000,
6
)
# 구조화된 로그 출력
log_entry = {
**request_data,
**response_data
}
if response.status_code >= 500:
logger.error(json.dumps(log_entry))
elif response.status_code >= 400:
logger.warning(json.dumps(log_entry))
else:
logger.info(json.dumps(log_entry))
# 응답 헤더에 request_id 추가
response.headers["X-Request-ID"] = request_id
return response
5단계: HolySheep AI 연동 클라이언트
# services/ai_client.py
import httpx
from typing import Optional, List, Dict, Any
from config import settings
import logging
logger = logging.getLogger(__name__)
class HolySheepAIClient:
"""
HolySheep AI 게이트웨이 클라이언트
HolySheep AI 특징:
- 단일 API 키로 다중 모델 지원 (GPT-4.1, Claude, Gemini, DeepSeek)
- 글로벌 리전 최적화 및 장애 자동 복구
- 투명한 비용 보고서 제공
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = settings.HOLYSHEEP_BASE_URL
self.client = httpx.AsyncClient(
base_url=self.base_url,
timeout=httpx.Timeout(60.0, connect=10.0),
headers={
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
)
async def chat_completion(
self,
messages: List[Dict[str, str]],
model: str = "gpt-4.1",
temperature: float = 0.7,
max_tokens: Optional[int] = None,