第9章：デプロイと運用

1. FastAPI

FastAPIは、モダンで高速なPython Webフレームワークである[1]。型ヒントを活用した自動ドキュメント生成と高いパフォーマンスが特徴。

1.1 基本構造

from fastapi import FastAPI, HTTPException, Depends, status
from fastapi.middleware.cors import CORSMiddleware
from pydantic import BaseModel, Field
from typing import Optional
import uvicorn

app = FastAPI(
    title="My API",
    description="サンプルAPI",
    version="1.0.0",
)

# CORS設定
app.add_middleware(
    CORSMiddleware,
    allow_origins=["https://example.com"],
    allow_credentials=True,
    allow_methods=["*"],
    allow_headers=["*"],
)

# リクエスト/レスポンスモデル
class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    price: float = Field(..., gt=0)
    description: Optional[str] = None

class ItemResponse(BaseModel):
    id: int
    name: str
    price: float
    description: Optional[str]

    class Config:
        from_attributes = True

# エンドポイント
@app.get("/")
async def root():
    return {"message": "Hello World"}

@app.post("/items", response_model=ItemResponse, status_code=status.HTTP_201_CREATED)
async def create_item(item: ItemCreate):
    # DB保存処理（省略）
    return ItemResponse(id=1, **item.model_dump())

@app.get("/items/{item_id}", response_model=ItemResponse)
async def get_item(item_id: int):
    # DB取得処理（省略）
    if item_id <= 0:
        raise HTTPException(status_code=404, detail="Item not found")
    return ItemResponse(id=item_id, name="Sample", price=100.0, description=None)

if __name__ == "__main__":
    uvicorn.run("main:app", host="0.0.0.0", port=8000, reload=True)

1.2 依存性注入

FastAPIの依存性注入システムにより、共通処理を再利用可能。

from fastapi import Depends, Header, HTTPException
from sqlalchemy.orm import Session
from typing import Annotated

# データベースセッション
def get_db():
    db = SessionLocal()
    try:
        yield db
    finally:
        db.close()

# 認証
async def verify_token(authorization: str = Header(...)):
    if not authorization.startswith("Bearer "):
        raise HTTPException(status_code=401, detail="Invalid token")
    token = authorization.replace("Bearer ", "")
    # トークン検証ロジック
    return {"user_id": 1, "role": "admin"}

# 型エイリアス（Python 3.9+）
DbSession = Annotated[Session, Depends(get_db)]
CurrentUser = Annotated[dict, Depends(verify_token)]

# エンドポイントで使用
@app.get("/protected")
async def protected_route(db: DbSession, user: CurrentUser):
    return {"user": user, "message": "Access granted"}

1.3 非同期処理とバックグラウンドタスク

from fastapi import BackgroundTasks
import asyncio
import httpx

# 非同期HTTP呼び出し
async def fetch_external_data(url: str) -> dict:
    async with httpx.AsyncClient() as client:
        response = await client.get(url)
        return response.json()

# バックグラウンドタスク
def send_notification(email: str, message: str):
    # メール送信処理（時間のかかる処理）
    print(f"Sending email to {email}: {message}")

@app.post("/orders")
async def create_order(
    order: OrderCreate,
    background_tasks: BackgroundTasks,
    db: DbSession,
):
    # 注文処理
    order_id = save_order(db, order)

    # バックグラウンドで通知送信
    background_tasks.add_task(
        send_notification,
        order.customer_email,
        f"Order {order_id} confirmed"
    )

    return {"order_id": order_id, "status": "processing"}

2. モデルサービング

機械学習モデルをAPIとして提供するパターンを解説する。

2.1 FastAPIでのモデルサービング

from fastapi import FastAPI
from pydantic import BaseModel
import torch
from transformers import AutoTokenizer, AutoModelForSequenceClassification
from contextlib import asynccontextmanager

# グローバル変数でモデルを保持
model = None
tokenizer = None

@asynccontextmanager
async def lifespan(app: FastAPI):
    # 起動時にモデルをロード
    global model, tokenizer
    model_name = "bert-base-uncased"
    tokenizer = AutoTokenizer.from_pretrained(model_name)
    model = AutoModelForSequenceClassification.from_pretrained(model_name)
    model.eval()
    print("Model loaded successfully")
    yield
    # シャットダウン時のクリーンアップ
    del model, tokenizer

app = FastAPI(lifespan=lifespan)

class PredictionRequest(BaseModel):
    text: str

class PredictionResponse(BaseModel):
    label: str
    confidence: float

@app.post("/predict", response_model=PredictionResponse)
async def predict(request: PredictionRequest):
    # 推論
    inputs = tokenizer(
        request.text,
        return_tensors="pt",
        truncation=True,
        max_length=512,
    )

    with torch.no_grad():
        outputs = model(**inputs)
        probs = torch.softmax(outputs.logits, dim=-1)
        confidence, predicted = torch.max(probs, dim=-1)

    labels = ["negative", "positive"]
    return PredictionResponse(
        label=labels[predicted.item()],
        confidence=confidence.item(),
    )

2.2 バッチ推論

複数リクエストをバッチ処理してスループット向上。

from fastapi import FastAPI
import asyncio
from collections import deque
from typing import List
import time

class BatchProcessor:
    def __init__(self, batch_size: int = 32, max_wait_ms: int = 100):
        self.batch_size = batch_size
        self.max_wait_ms = max_wait_ms
        self.queue = deque()
        self.lock = asyncio.Lock()

    async def add_request(self, data: dict) -> dict:
        future = asyncio.Future()
        async with self.lock:
            self.queue.append((data, future))

            if len(self.queue) >= self.batch_size:
                await self._process_batch()

        # タイムアウト待機
        try:
            return await asyncio.wait_for(future, timeout=self.max_wait_ms / 1000 + 5)
        except asyncio.TimeoutError:
            return {"error": "Timeout"}

    async def _process_batch(self):
        if not self.queue:
            return

        batch = []
        futures = []
        while self.queue and len(batch) < self.batch_size:
            data, future = self.queue.popleft()
            batch.append(data)
            futures.append(future)

        # バッチ推論実行
        results = await self._run_inference(batch)

        # 結果を各futureに設定
        for future, result in zip(futures, results):
            future.set_result(result)

    async def _run_inference(self, batch: List[dict]) -> List[dict]:
        # 実際のモデル推論
        # ...
        return [{"prediction": "result"} for _ in batch]

batch_processor = BatchProcessor()

@app.post("/predict/batch")
async def predict_batch(request: PredictionRequest):
    return await batch_processor.add_request({"text": request.text})

Table 1. モデルサービング手法の比較

3. コンテナ化

Dockerによるコンテナ化で、環境の再現性と可搬性を確保する[2]。

3.1 本番用Dockerfile

# Dockerfile
FROM python:3.12-slim as builder

WORKDIR /app

# 依存関係のインストール
RUN pip install uv
COPY pyproject.toml uv.lock ./
RUN uv sync --frozen --no-dev

FROM python:3.12-slim as runtime

WORKDIR /app

# 非rootユーザー作成
RUN useradd --create-home --shell /bin/bash appuser

# ビルドステージから仮想環境をコピー
COPY --from=builder /app/.venv /app/.venv
ENV PATH="/app/.venv/bin:$PATH"

# アプリケーションコード
COPY --chown=appuser:appuser ./src ./src

USER appuser

# ヘルスチェック
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
    CMD curl -f http://localhost:8000/health || exit 1

EXPOSE 8000

CMD ["uvicorn", "src.main:app", "--host", "0.0.0.0", "--port", "8000"]

3.2 docker-compose

# docker-compose.yml
version: '3.8'

services:
  api:
    build:
      context: .
      dockerfile: Dockerfile
    ports:
      - "8000:8000"
    environment:
      - DATABASE_URL=postgresql://user:pass@db:5432/mydb
      - REDIS_URL=redis://redis:6379/0
    depends_on:
      db:
        condition: service_healthy
      redis:
        condition: service_started
    deploy:
      resources:
        limits:
          cpus: '2'
          memory: 2G
    restart: unless-stopped

  db:
    image: postgres:16
    environment:
      POSTGRES_USER: user
      POSTGRES_PASSWORD: pass
      POSTGRES_DB: mydb
    volumes:
      - postgres_data:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U user -d mydb"]
      interval: 10s
      timeout: 5s
      retries: 5

  redis:
    image: redis:7-alpine
    volumes:
      - redis_data:/data

  nginx:
    image: nginx:alpine
    ports:
      - "80:80"
      - "443:443"
    volumes:
      - ./nginx.conf:/etc/nginx/nginx.conf:ro
      - ./certs:/etc/nginx/certs:ro
    depends_on:
      - api

volumes:
  postgres_data:
  redis_data:

Fig. 1にデプロイアーキテクチャを示す。

4. 監視とロギング

4.1 構造化ロギング

JSON形式のログにより、ログ集約・分析が容易になる。

import logging
import json
from datetime import datetime
from typing import Any

class JSONFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        log_data = {
            "timestamp": datetime.utcnow().isoformat(),
            "level": record.levelname,
            "message": record.getMessage(),
            "logger": record.name,
            "module": record.module,
            "function": record.funcName,
            "line": record.lineno,
        }

        # 追加のコンテキスト
        if hasattr(record, 'request_id'):
            log_data['request_id'] = record.request_id
        if hasattr(record, 'user_id'):
            log_data['user_id'] = record.user_id

        # 例外情報
        if record.exc_info:
            log_data['exception'] = self.formatException(record.exc_info)

        return json.dumps(log_data)

# ロガー設定
def setup_logging():
    handler = logging.StreamHandler()
    handler.setFormatter(JSONFormatter())

    logger = logging.getLogger()
    logger.addHandler(handler)
    logger.setLevel(logging.INFO)

# FastAPIミドルウェアでリクエストID追加
from fastapi import Request
import uuid

@app.middleware("http")
async def add_request_id(request: Request, call_next):
    request_id = str(uuid.uuid4())
    request.state.request_id = request_id

    # ログにリクエストIDを追加
    logger = logging.getLogger()
    old_factory = logging.getLogRecordFactory()

    def record_factory(*args, **kwargs):
        record = old_factory(*args, **kwargs)
        record.request_id = request_id
        return record

    logging.setLogRecordFactory(record_factory)
    response = await call_next(request)
    response.headers["X-Request-ID"] = request_id
    return response

4.2 Prometheus メトリクス

アプリケーションメトリクスをPrometheusで収集[3]。

from prometheus_client import Counter, Histogram, generate_latest, CONTENT_TYPE_LATEST
from fastapi import FastAPI, Request, Response
import time

# メトリクス定義
REQUEST_COUNT = Counter(
    'http_requests_total',
    'Total HTTP requests',
    ['method', 'endpoint', 'status']
)

REQUEST_LATENCY = Histogram(
    'http_request_duration_seconds',
    'HTTP request latency',
    ['method', 'endpoint'],
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 5.0]
)

MODEL_INFERENCE_TIME = Histogram(
    'model_inference_seconds',
    'Model inference time',
    ['model_name'],
    buckets=[0.01, 0.05, 0.1, 0.5, 1.0, 2.0, 5.0]
)

# ミドルウェア
@app.middleware("http")
async def metrics_middleware(request: Request, call_next):
    start_time = time.time()
    response = await call_next(request)
    duration = time.time() - start_time

    REQUEST_COUNT.labels(
        method=request.method,
        endpoint=request.url.path,
        status=response.status_code
    ).inc()

    REQUEST_LATENCY.labels(
        method=request.method,
        endpoint=request.url.path
    ).observe(duration)

    return response

# メトリクスエンドポイント
@app.get("/metrics")
async def metrics():
    return Response(
        content=generate_latest(),
        media_type=CONTENT_TYPE_LATEST
    )

Table 2. 監視すべき主要メトリクス

5. セキュリティ

5.1 認証と認可

from fastapi import Depends, HTTPException, status
from fastapi.security import OAuth2PasswordBearer, OAuth2PasswordRequestForm
from jose import JWTError, jwt
from passlib.context import CryptContext
from datetime import datetime, timedelta
from pydantic import BaseModel

# 設定
SECRET_KEY = "your-secret-key"  # 環境変数から取得
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30

pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto")
oauth2_scheme = OAuth2PasswordBearer(tokenUrl="token")

class Token(BaseModel):
    access_token: str
    token_type: str

class TokenData(BaseModel):
    username: str | None = None
    scopes: list[str] = []

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

async def get_current_user(token: str = Depends(oauth2_scheme)):
    credentials_exception = HTTPException(
        status_code=status.HTTP_401_UNAUTHORIZED,
        detail="Could not validate credentials",
        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
        return TokenData(username=username, scopes=payload.get("scopes", []))
    except JWTError:
        raise credentials_exception

# ロールベースアクセス制御
def require_role(required_role: str):
    async def role_checker(user: TokenData = Depends(get_current_user)):
        if required_role not in user.scopes:
            raise HTTPException(
                status_code=status.HTTP_403_FORBIDDEN,
                detail="Insufficient permissions"
            )
        return user
    return role_checker

@app.get("/admin")
async def admin_only(user: TokenData = Depends(require_role("admin"))):
    return {"message": f"Welcome admin {user.username}"}

5.2 入力バリデーションとサニタイズ

from pydantic import BaseModel, field_validator, Field
from typing import Annotated
import re
import bleach

class UserInput(BaseModel):
    username: str = Field(..., min_length=3, max_length=50)
    email: str
    bio: str = Field(..., max_length=1000)

    @field_validator('username')
    @classmethod
    def username_alphanumeric(cls, v):
        if not re.match(r'^[a-zA-Z0-9_]+$', v):
            raise ValueError('Username must be alphanumeric')
        return v

    @field_validator('email')
    @classmethod
    def email_valid(cls, v):
        pattern = r'^[a-zA-Z0-9._%+-]+@[a-zA-Z0-9.-]+\.[a-zA-Z]{2,}$'
        if not re.match(pattern, v):
            raise ValueError('Invalid email format')
        return v.lower()

    @field_validator('bio')
    @classmethod
    def sanitize_bio(cls, v):
        # HTMLタグを除去
        return bleach.clean(v, tags=[], strip=True)

5.3 セキュリティヘッダー

from fastapi import FastAPI
from fastapi.middleware.trustedhost import TrustedHostMiddleware
from starlette.middleware.httpsredirect import HTTPSRedirectMiddleware

app = FastAPI()

# HTTPS強制
app.add_middleware(HTTPSRedirectMiddleware)

# 信頼するホスト
app.add_middleware(
    TrustedHostMiddleware,
    allowed_hosts=["example.com", "*.example.com"]
)

# セキュリティヘッダー
@app.middleware("http")
async def add_security_headers(request, call_next):
    response = await call_next(request)
    response.headers["X-Content-Type-Options"] = "nosniff"
    response.headers["X-Frame-Options"] = "DENY"
    response.headers["X-XSS-Protection"] = "1; mode=block"
    response.headers["Strict-Transport-Security"] = "max-age=31536000; includeSubDomains"
    response.headers["Content-Security-Policy"] = "default-src 'self'"
    return response

Fig. 2にセキュリティレイヤーを示す。

Table 3. セキュリティチェックリスト

References

[1] FastAPI, "FastAPI Documentation," fastapi.tiangolo.com, 2024.

[2] Docker, "Docker Documentation," docs.docker.com, 2024.

[3] Prometheus, "Prometheus Documentation," prometheus.io/docs, 2024.

[4] OWASP, "OWASP Top Ten," owasp.org, 2024.