BehavioralPythonverifiedVerified

Chain of Responsibility Pattern in Python

Passes a request along a chain of handlers, where each handler decides to process it or pass it to the next handler in the chain.

How to Implement the Chain of Responsibility Pattern in Python

1Step 1: Define the request type and abstract handler

from __future__ import annotations
from dataclasses import dataclass
from typing import Any


@dataclass
class Request:
    type: str
    payload: Any


class Handler:
    def __init__(self) -> None:
        self._next: Handler | None = None

    def set_next(self, handler: Handler) -> Handler:
        self._next = handler
        return handler

    def handle(self, request: Request) -> str | None:
        if self._next:
            return self._next.handle(request)
        return None

2Step 2: Implement concrete handlers

class AuthHandler(Handler):
    def handle(self, request: Request) -> str | None:
        if request.type == "auth":
            return "Handled by AuthHandler"
        return super().handle(request)


class ValidationHandler(Handler):
    def handle(self, request: Request) -> str | None:
        if request.type == "validate":
            return "Handled by ValidationHandler"
        return super().handle(request)


class LoggingHandler(Handler):
    def handle(self, request: Request) -> str | None:
        print(f"Logging: {request.type}")
        return super().handle(request)

3Step 3: Build the chain and process requests

auth = AuthHandler()
validation = ValidationHandler()
logging_handler = LoggingHandler()

logging_handler.set_next(auth).set_next(validation)

logging_handler.handle(Request("auth", {}))
logging_handler.handle(Request("validate", {}))

"""HTTP Middleware Chain (ASGI-style) using Chain of Responsibility."""

from dataclasses import dataclass, field
from typing import Any, Callable, Awaitable


# [step] Define HTTP request/response types and middleware signature
@dataclass
class HttpRequest:
    method: str
    path: str
    headers: dict[str, str]
    body: Any
    context: dict[str, Any] = field(default_factory=dict)


@dataclass
class HttpResponse:
    status: int
    headers: dict[str, str]
    body: Any


NextFunction = Callable[[], Awaitable[HttpResponse]]
Middleware = Callable[[HttpRequest, NextFunction], Awaitable[HttpResponse]]


# [step] Implement concrete middleware handlers
async def cors_middleware(req: HttpRequest, next_fn: NextFunction) -> HttpResponse:
    response = await next_fn()
    response.headers.update({
        "Access-Control-Allow-Origin": "*",
        "Access-Control-Allow-Methods": "GET, POST, PUT, DELETE",
    })
    return response


async def auth_middleware(req: HttpRequest, next_fn: NextFunction) -> HttpResponse:
    token = req.headers.get("authorization", "")
    if not token.startswith("Bearer "):
        return HttpResponse(401, {}, {"error": "Unauthorized"})
    req.context["user_id"] = "user_123"
    return await next_fn()


async def rate_limit_middleware(req: HttpRequest, next_fn: NextFunction) -> HttpResponse:
    client_ip = req.headers.get("x-forwarded-for", "unknown")
    key = f"rate:{client_ip}"
    requests = req.context.get(key, 0)
    if requests > 100:
        return HttpResponse(429, {"Retry-After": "60"}, {"error": "Too many requests"})
    req.context[key] = requests + 1
    return await next_fn()


# [step] Build the pipeline that chains middleware together
class MiddlewarePipeline:
    def __init__(self) -> None:
        self._middlewares: list[Middleware] = []

    def use(self, middleware: Middleware) -> "MiddlewarePipeline":
        self._middlewares.append(middleware)
        return self

    async def execute(
        self,
        req: HttpRequest,
        handler: Callable[[HttpRequest], Awaitable[HttpResponse]],
    ) -> HttpResponse:
        index = 0
        middlewares = self._middlewares

        async def next_fn() -> HttpResponse:
            nonlocal index
            if index < len(middlewares):
                mw = middlewares[index]
                index += 1
                return await mw(req, next_fn)
            return await handler(req)

        return await next_fn()


# [step] Assemble and use the pipeline
pipeline = (
    MiddlewarePipeline()
    .use(cors_middleware)
    .use(rate_limit_middleware)
    .use(auth_middleware)
)

Chain of Responsibility Pattern Architecture

hourglass_empty

Rendering diagram...

lightbulb

Chain of Responsibility Pattern in the Real World

“Like a customer support escalation: your call starts with a front-line agent. If they can’t resolve it, they transfer you to a specialist. If the specialist can’t help, it goes to a manager. Each level either handles it or passes it up the chain.”