ミドルウェアチェーンを実装パターンとして徹底理解する

def handle_users(request):
    # ログ記録（コピペ 1/3）
    print(f"{request.method} {request.path}")
    # 認証チェック（コピペ 1/3）
    if "Authorization" not in request.headers:
        return Response(status=401, body="Unauthorized")
    # エラーハンドリング（コピペ 1/3）
    try:
        return Response(status=200, body="User list")
    except Exception:
        return Response(status=500, body="Internal Server Error")
 
 
def handle_orders(request):
    # ログ記録（コピペ 2/3）
    print(f"{request.method} {request.path}")
    # 認証チェック（コピペ 2/3）
    if "Authorization" not in request.headers:
        return Response(status=401, body="Unauthorized")
    # エラーハンドリング（コピペ 2/3）
    try:
        return Response(status=200, body="Order list")
    except Exception:
        return Response(status=500, body="Internal Server Error")
 
 
def handle_products(request):
    # ログ記録（コピペ 3/3）
    print(f"{request.method} {request.path}")
    # 認証チェック（コピペ 3/3）
    if "Authorization" not in request.headers:
        return Response(status=401, body="Unauthorized")
    # エラーハンドリング（コピペ 3/3）
    try:
        return Response(status=200, body="Product list")
    except Exception:
        return Response(status=500, body="Internal Server Error")

# ミドルウェアチェーンがあれば、ハンドラーはビジネスロジックだけ
def handle_users(request):
    return Response(status=200, body="User list")
 
def handle_orders(request):
    return Response(status=200, body="Order list")
 
def handle_products(request):
    return Response(status=200, body="Product list")
 
# 横断的関心事はミドルウェアとして1箇所で定義
app = compose(router, [error_handling, logging, auth])

from __future__ import annotations
 
from dataclasses import dataclass, field
from collections.abc import Callable
 
 
# ──────────────────────────────────────────────
# リクエストとレスポンスの型定義
# ──────────────────────────────────────────────
 
@dataclass
class Request:
    """HTTP リクエストを模倣するシンプルなデータクラス。"""
    method: str = "GET"
    path: str = "/"
    headers: dict[str, str] = field(default_factory=dict)
    body: str = ""
 
 
@dataclass
class Response:
    """HTTP レスポンスを模倣するシンプルなデータクラス。"""
    status: int = 200
    headers: dict[str, str] = field(default_factory=dict)
    body: str = ""
 
 
# ──────────────────────────────────────────────
# 型エイリアス
# ──────────────────────────────────────────────
 
# Handler: リクエストを受け取りレスポンスを返す関数
Handler = Callable[[Request], Response]
 
# Middleware: 次のハンドラーを受け取り、新しいハンドラーを返す関数
Middleware = Callable[[Handler], Handler]

def logging_middleware(next_handler: Handler) -> Handler:
    """リクエストとレスポンスをログ出力するミドルウェア。"""
    def handler(request: Request) -> Response:
        # ── 前処理（リクエストがハンドラーに届く前） ──
        print(f"--> {request.method} {request.path}")
 
        # ── next を呼ぶ ──
        response = next_handler(request)
 
        # ── 後処理（レスポンスがクライアントに返る前） ──
        print(f"<-- {response.status}")
 
        return response
 
    return handler

def logging_middleware(next_handler: Handler) -> Handler:
    #           ↑ next_handler はここでローカルスコープに入る
 
    def handler(request: Request) -> Response:
        #     ↑ handler は next_handler を「キャプチャ」している
        print(f"--> {request.method} {request.path}")
        response = next_handler(request)  # ← クロージャ経由でアクセス
        print(f"<-- {response.status}")
        return response
 
    return handler
    # ← logging_middleware のスタックフレームが終了しても、
    #    handler が next_handler への参照を保持し続ける

def compose(handler: Handler, middlewares: list[Middleware]) -> Handler:
    """ミドルウェアのリストを合成して1つのハンドラーにする。
 
    middlewares = [A, B, C] の場合、実行順序は A -> B -> C -> handler となる。
    内部的には handler を C で包み、それを B で包み、さらに A で包む。
    """
    composed = handler
    for middleware in reversed(middlewares):
        composed = middleware(composed)
    return composed

compose 実行時（起動前の配線フェーズ）:
  composed = handler
  composed = auth(handler)           → auth_handler   (handler をクロージャで保持)
  composed = timing(auth_handler)    → timing_handler  (auth_handler をクロージャで保持)
  composed = logging(timing_handler) → logging_handler (timing_handler をクロージャで保持)
 
app = logging_handler  ← これが最終的なエントリーポイント

app(Request("GET /")) 呼び出し時（実行フェーズ）:
 
  logging_handler(request)
  │  print("--> GET /")                      # logging 前処理
  │
  ├──▶ timing_handler(request)               # next_handler 呼び出し
  │    │  start = time.perf_counter()        # timing 前処理
  │    │
  │    ├──▶ auth_handler(request)            # next_handler 呼び出し
  │    │    │  Authorization ヘッダー確認     # auth 前処理
  │    │    │
  │    │    ├──▶ handler(request)            # next_handler 呼び出し
  │    │    │    │  return Response(200)     # ビジネスロジック実行
  │    │    │    ▼
  │    │    │  response = Response(200)
  │    │    │  return response               # auth 後処理なし → そのまま返す
  │    │    ▼
  │    │  elapsed_ms = ...                   # timing 後処理
  │    │  response.headers["X-Response-Time"] = "1.23ms"
  │    │  return response
  │    ▼
  │  print("<-- 200")                        # logging 後処理
  │  return response
  ▼

def auth_middleware(next_handler: Handler) -> Handler:
    """認証チェックを行うミドルウェア。
 
    Authorization ヘッダーが無ければ 401 を返す（ショートサーキット）。
    """
    def handler(request: Request) -> Response:
        if "Authorization" not in request.headers:
            # next_handler を呼ばずにレスポンスを返す → ショートサーキット
            return Response(status=401, body="Unauthorized")
 
        return next_handler(request)
 
    return handler
 
 
def timing_middleware(next_handler: Handler) -> Handler:
    """処理時間を計測するミドルウェア。"""
    import time
 
    def handler(request: Request) -> Response:
        start = time.perf_counter()
        response = next_handler(request)
        elapsed_ms = (time.perf_counter() - start) * 1000
        response.headers["X-Response-Time"] = f"{elapsed_ms:.2f}ms"
        return response
 
    return handler
 
 
# ── 最終ハンドラー ──
def my_handler(request: Request) -> Response:
    return Response(status=200, body="Hello, World!")
 
 
# ── チェーンの組み立てと実行 ──
app = compose(my_handler, [logging_middleware, timing_middleware, auth_middleware])
 
# 認証ヘッダーありのリクエスト
print("=== With Auth ===")
resp = app(Request(method="GET", path="/", headers={"Authorization": "Bearer token123"}))
print(f"Body: {resp.body}\n")
 
# 認証ヘッダーなしのリクエスト（ショートサーキットされる）
print("=== Without Auth ===")
resp = app(Request(method="GET", path="/"))
print(f"Status: {resp.status}, Body: {resp.body}")

=== With Auth ===
--> GET /
<-- 200
Body: Hello, World!
 
=== Without Auth ===
--> GET /
<-- 401
Status: 401, Body: Unauthorized

class CacheMiddleware:
    """レスポンスをキャッシュし、同じリクエストにはキャッシュから応答する。
 
    キャッシュヒット時はパイプラインの残りを完全にスキップする。
    これにより DB アクセスや外部 API 呼び出しを回避し、
    レスポンス時間を大幅に短縮できる。
    """
 
    def __init__(self) -> None:
        self._cache: dict[str, Response] = {}
 
    def __call__(self, next_handler: Handler) -> Handler:
        def handler(request: Request) -> Response:
            cache_key = f"{request.method}:{request.path}"
 
            # キャッシュヒット → ショートサーキット
            if cache_key in self._cache:
                cached = self._cache[cache_key]
                cached.headers["X-Cache"] = "HIT"
                return cached  # next_handler を呼ばない！
 
            # キャッシュミス → パイプラインを続行
            response = next_handler(request)
 
            # 成功レスポンスのみキャッシュ
            if response.status == 200:
                response.headers["X-Cache"] = "MISS"
                self._cache[cache_key] = response
 
            return response
 
        return handler

def error_handling_middleware(next_handler: Handler) -> Handler:
    """例外を捕捉して 500 レスポンスに変換するミドルウェア。"""
 
    def handler(request: Request) -> Response:
        try:
            return next_handler(request)
        except Exception as exc:
            print(f"[ERROR] {type(exc).__name__}: {exc}")
            return Response(status=500, body="Internal Server Error")
 
    return handler

// Express.js のエラーハンドリングミドルウェア
// 引数が4つ（err を含む）であることが識別子
app.use((err, req, res, next) => {
  console.error(err.stack)
  res.status(500).send('Something broke!')
})

# 危険な順序：認証がエラーハンドリングの外側
bad_pipeline = compose(my_handler, [
    auth_middleware,              # ← 最外層
    error_handling_middleware,    # ← 内側
    timing_middleware,
])

# 正しい順序：エラーハンドリングが最外層
good_pipeline = compose(my_handler, [
    error_handling_middleware,    # ← 最外層：全例外を捕捉
    auth_middleware,
    timing_middleware,
])

# 問題のある順序
bad_pipeline = compose(my_handler, [
    auth_middleware,
    logging_middleware,   # ← 認証の内側
])

# 正しい順序
good_pipeline = compose(my_handler, [
    logging_middleware,   # ← 最外層：全リクエストを記録
    auth_middleware,
])

def make_trace_middleware(name: str) -> Middleware:
    """実行順序を確認するためのトレースミドルウェアを生成する。"""
 
    def middleware(next_handler: Handler) -> Handler:
        def handler(request: Request) -> Response:
            print(f"  [{name}] 前処理")
            response = next_handler(request)
            print(f"  [{name}] 後処理")
            return response
        return handler
 
    return middleware
 
 
def trace_handler(request: Request) -> Response:
    print(f"  [handler] 実行")
    return Response(status=200, body="OK")
 
 
# ── A -> B -> C -> handler の順序で合成 ──
pipeline = compose(trace_handler, [
    make_trace_middleware("A"),
    make_trace_middleware("B"),
    make_trace_middleware("C"),
])
 
print("実行順序:")
pipeline(Request())

実行順序:
  [A] 前処理
  [B] 前処理
  [C] 前処理
  [handler] 実行
  [C] 後処理
  [B] 後処理
  [A] 後処理

const express = require('express')
const app = express()
 
// ミドルウェア 1: ログ記録
app.use((req, res, next) => {
  console.log(`${req.method} ${req.url}`)
  next()  // 次のミドルウェアへ
})
 
// ミドルウェア 2: 認証
app.use((req, res, next) => {
  if (!req.headers.authorization) {
    return res.status(401).send('Unauthorized') // ショートサーキット
  }
  next()
})
 
// ハンドラー
app.get('/', (req, res) => {
  res.send('Hello!')
})

# Django のミドルウェア（関数スタイル）
def simple_middleware(get_response):
    # 1回だけ実行される初期化
 
    def middleware(request):
        # 前処理
        response = get_response(request)
        # 後処理
        return response
 
    return middleware
 
 
# Django のミドルウェア（クラススタイル）
class SimpleMiddleware:
    def __init__(self, get_response):
        self.get_response = get_response
 
    def __call__(self, request):
        # 前処理
        response = self.get_response(request)
        # 後処理
        return response

var app = builder.Build();
 
// Use(): 次のミドルウェアを呼ぶことができる
app.Use(async (context, next) =>
{
    Console.WriteLine("Middleware 1: before");
    await next.Invoke(context);
    Console.WriteLine("Middleware 1: after");
});
 
app.Use(async (context, next) =>
{
    Console.WriteLine("Middleware 2: before");
    await next.Invoke(context);
    Console.WriteLine("Middleware 2: after");
});
 
// Run(): ターミナルミドルウェア（next を受け取らない）
app.Run(async context =>
{
    await context.Response.WriteAsync("Hello!");
});

Middleware 1: before
Middleware 2: before
Middleware 2: after
Middleware 1: after

func loggingMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        log.Printf("%s %s", r.Method, r.URL.Path)
        next.ServeHTTP(w, r)  // 次のハンドラーへ
    })
}
 
func authMiddleware(next http.Handler) http.Handler {
    return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) {
        if r.Header.Get("Authorization") == "" {
            http.Error(w, "Unauthorized", http.StatusUnauthorized)
            return // ショートサーキット
        }
        next.ServeHTTP(w, r)
    })
}
 
// チェーンの組み立て
handler := loggingMiddleware(authMiddleware(myHandler))

def test_auth_middleware_rejects_unauthenticated() -> None:
    """認証ヘッダーがないリクエストは 401 になることをテスト。"""
    # ── モックハンドラー（呼ばれないはず） ──
    called = False
 
    def mock_handler(request: Request) -> Response:
        nonlocal called
        called = True
        return Response(status=200, body="OK")
 
    # ── テスト対象のミドルウェアを適用 ──
    protected = auth_middleware(mock_handler)
 
    # ── 認証なしリクエストでテスト ──
    response = protected(Request(method="GET", path="/secret"))
 
    assert response.status == 401
    assert response.body == "Unauthorized"
    assert not called, "ハンドラーは呼ばれないはず"
 
 
def test_auth_middleware_passes_authenticated() -> None:
    """認証ヘッダーがあるリクエストはハンドラーまで到達することをテスト。"""
    def mock_handler(request: Request) -> Response:
        return Response(status=200, body="Secret Content")
 
    protected = auth_middleware(mock_handler)
 
    response = protected(Request(
        method="GET",
        path="/secret",
        headers={"Authorization": "Bearer token123"},
    ))
 
    assert response.status == 200
    assert response.body == "Secret Content"
 
 
def test_compose_order() -> None:
    """compose が登録順で前処理、逆順で後処理を実行することをテスト。"""
    order: list[str] = []
 
    def make_tracker(name: str) -> Middleware:
        def middleware(next_handler: Handler) -> Handler:
            def handler(request: Request) -> Response:
                order.append(f"{name}:pre")
                response = next_handler(request)
                order.append(f"{name}:post")
                return response
            return handler
        return middleware
 
    def final_handler(request: Request) -> Response:
        order.append("handler")
        return Response(status=200)
 
    pipeline = compose(final_handler, [make_tracker("A"), make_tracker("B")])
    pipeline(Request())
 
    assert order == ["A:pre", "B:pre", "handler", "B:post", "A:post"]

class RateLimitMiddleware:
    """一定時間内のリクエスト数を制限するミドルウェア。
 
    クラスベースにすることで、リクエスト数のカウンターを
    インスタンス変数として保持できる。
    """
 
    def __init__(self, max_requests: int = 100, window_seconds: float = 60.0) -> None:
        self.max_requests = max_requests
        self.window_seconds = window_seconds
        self._request_counts: dict[str, list[float]] = {}
 
    def __call__(self, next_handler: Handler) -> Handler:
        import time
 
        def handler(request: Request) -> Response:
            client_ip = request.headers.get("X-Forwarded-For", "unknown")
            now = time.time()
 
            # ウィンドウ外の古い記録を除去
            timestamps = self._request_counts.get(client_ip, [])
            timestamps = [t for t in timestamps if now - t < self.window_seconds]
 
            if len(timestamps) >= self.max_requests:
                return Response(status=429, body="Too Many Requests")
 
            timestamps.append(now)
            self._request_counts[client_ip] = timestamps
 
            return next_handler(request)
 
        return handler
 
 
# クラスベースミドルウェアの使用例
rate_limiter = RateLimitMiddleware(max_requests=10, window_seconds=60.0)
app = compose(my_handler, [logging_middleware, rate_limiter, auth_middleware])