Chain of Responsibility и Middleware

Цепочка превращает множество обработчиков в единый конвейер.

#Chain of Responsibility: цепочка обработчиков

Chain of Responsibility — поведенческий паттерн, передающий запрос по цепочке обработчиков до тех пор, пока один из них не обработает запрос.

#Проблема: множество условий

# ❌ ПЛОХО: Большая цепочка if/elif
def handle_request(request: dict):
    if request.get("type") == "auth":
        return handle_auth(request)
    elif request.get("type") == "payment":
        return handle_payment(request)
    elif request.get("type") == "notification":
        return handle_notification(request)
    elif request.get("type") == "logging":
        return handle_logging(request)
    # ... ещё 10 условий ...
    else:
        raise ValueError("Unknown type")

# Проблема:
# 1. Нарушен Open-Closed — добавление требует изменения функции
# 2. Все обработчики в одной функции
# 3. Трудно тестировать отдельные обработчики

#Решение 1: Цепочка через классы

# ✅ ХОРОШО: Каждый обработчик — отдельный класс
from abc import ABC, abstractmethod
from typing import Optional

class Handler(ABC):
    def __init__(self):
        self._next: Optional["Handler"] = None
    
    def set_next(self, handler: "Handler") -> "Handler":
        self._next = handler
        return handler  # Возвращаем для цепочки
    
    @abstractmethod
    def handle(self, request: dict) -> Optional[dict]:
        pass
    
    def _handle_next(self, request: dict) -> Optional[dict]:
        if self._next:
            return self._next.handle(request)
        return None

class AuthHandler(Handler):
    def handle(self, request: dict) -> Optional[dict]:
        if request.get("type") == "auth":
            print("Handling auth")
            return {"status": "authenticated"}
        return self._handle_next(request)

class PaymentHandler(Handler):
    def handle(self, request: dict) -> Optional[dict]:
        if request.get("type") == "payment":
            print("Handling payment")
            return {"status": "paid"}
        return self._handle_next(request)

class NotificationHandler(Handler):
    def handle(self, request: dict) -> Optional[dict]:
        if request.get("type") == "notification":
            print("Handling notification")
            return {"status": "sent"}
        return self._handle_next(request)

# Использование
auth = AuthHandler()
payment = PaymentHandler()
notification = NotificationHandler()

# Строим цепочку
auth.set_next(payment).set_next(notification)

# Запрос проходит по цепочке
result = auth.handle({"type": "payment"})
# Handling payment
print(result)  # {"status": "paid"}

result = auth.handle({"type": "unknown"})
# None — ни один обработчик не справился

#Решение 2: Цепочка через список

# ✅ Проще: цепочка через список
from typing import List, Callable, Optional

RequestHandler = Callable[[dict], Optional[dict]]

class Chain:
    def __init__(self):
        self._handlers: List[RequestHandler] = []
    
    def add(self, handler: RequestHandler) -> "Chain":
        self._handlers.append(handler)
        return self
    
    def handle(self, request: dict) -> Optional[dict]:
        for handler in self._handlers:
            result = handler(request)
            if result is not None:
                return result
        return None

# Обработчики как функции
def auth_handler(request: dict) -> Optional[dict]:
    if request.get("type") == "auth":
        return {"status": "authenticated"}
    return None

def payment_handler(request: dict) -> Optional[dict]:
    if request.get("type") == "payment":
        return {"status": "paid"}
    return None

# Использование
chain = Chain()
chain.add(auth_handler).add(payment_handler)

result = chain.handle({"type": "payment"})
print(result)  # {"status": "paid"}

#Middleware: Chain of Responsibility в веб-фреймворках

#Middleware для HTTP-запросов

# ✅ Middleware pattern
from typing import Callable, List
from functools import wraps

Request = dict
Response = dict
Handler = Callable[[Request], Response]
Middleware = Callable[[Handler], Handler]

class MiddlewareChain:
    def __init__(self):
        self._middlewares: List[Middleware] = []
    
    def add(self, middleware: Middleware) -> "MiddlewareChain":
        self._middlewares.append(middleware)
        return self
    
    def build(self, final_handler: Handler) -> Handler:
        """Строит цепочку middleware"""
        handler = final_handler
        # Применяем middleware в обратном порядке
        for middleware in reversed(self._middlewares):
            handler = middleware(handler)
        return handler

# Middleware
def logging_middleware(handler: Handler) -> Handler:
    @wraps(handler)
    def wrapper(request: Request) -> Response:
        print(f"Request: {request}")
        response = handler(request)
        print(f"Response: {response}")
        return response
    return wrapper

def auth_middleware(handler: Handler) -> Handler:
    @wraps(handler)
    def wrapper(request: Request) -> Response:
        if not request.get("authorized"):
            return {"status": 401, "body": "Unauthorized"}
        return handler(request)
    return wrapper

def cors_middleware(handler: Handler) -> Handler:
    @wraps(handler)
    def wrapper(request: Request) -> Response:
        response = handler(request)
        response["headers"] = response.get("headers", {})
        response["headers"]["Access-Control-Allow-Origin"] = "*"
        return response
    return wrapper

# Финальный обработчик
def final_handler(request: Request) -> Response:
    return {"status": 200, "body": "OK"}

# Использование
chain = MiddlewareChain()
chain.add(logging_middleware)
chain.add(auth_middleware)
chain.add(cors_middleware)

handler = chain.build(final_handler)

# Запрос
response = handler({"authorized": True, "path": "/api"})
# Request: {...}
# Response: {...}

#Short-circuit в middleware

# ✅ Middleware может прервать цепочку
def rate_limit_middleware(handler: Handler) -> Handler:
    requests_count = {}
    
    @wraps(handler)
    def wrapper(request: Request) -> Response:
        ip = request.get("ip", "unknown")
        requests_count[ip] = requests_count.get(ip, 0) + 1
        
        if requests_count[ip] > 100:
            return {"status": 429, "body": "Too Many Requests"}
        
        return handler(request)
    return wrapper

# Rate limit срабатывает до основного обработчика

#Практическое применение

#Валидация данных

from typing import List, Tuple, Optional
from dataclasses import dataclass

@dataclass
class ValidationError:
    field: str
    message: str

Validator = Callable[[dict], List[ValidationError]]

class ValidationChain:
    def __init__(self):
        self._validators: List[Validator] = []
    
    def add(self, validator: Validator) -> "ValidationChain":
        self._validators.append(validator)
        return self
    
    def validate(self, data: dict) -> List[ValidationError]:
        errors = []
        for validator in self._validators:
            errors.extend(validator(data))
        return errors

# Валидаторы
def validate_email(data: dict) -> List[ValidationError]:
    errors = []
    email = data.get("email", "")
    if not email:
        errors.append(ValidationError("email", "Email required"))
    elif "@" not in email:
        errors.append(ValidationError("email", "Invalid email format"))
    return errors

def validate_password(data: dict) -> List[ValidationError]:
    errors = []
    password = data.get("password", "")
    if len(password) < 8:
        errors.append(ValidationError("password", "Password too short"))
    if not any(c.isupper() for c in password):
        errors.append(ValidationError("password", "Password needs uppercase"))
    return errors

def validate_age(data: dict) -> List[ValidationError]:
    errors = []
    age = data.get("age")
    if age is None:
        errors.append(ValidationError("age", "Age required"))
    elif age < 18:
        errors.append(ValidationError("age", "Must be 18+"))
    return errors

# Использование
chain = ValidationChain()
chain.add(validate_email).add(validate_password).add(validate_age)

errors = chain.validate({"email": "invalid", "password": "123", "age": 15})
for error in errors:
    print(f"{error.field}: {error.message}")
# email: Invalid email format
# password: Password too short
# password: Password needs uppercase
# age: Must be 18+

#Обработка исключений в цепочке

from contextlib import contextmanager

ErrorHandler = Callable[[Exception], Optional[Response]]

class ExceptionHandlerChain:
    def __init__(self):
        self._handlers: List[ErrorHandler] = []
    
    def add(self, handler: ErrorHandler) -> "ExceptionHandlerChain":
        self._handlers.append(handler)
        return self
    
    def handle(self, exc: Exception) -> Optional[Response]:
        for handler in self._handlers:
            result = handler(exc)
            if result is not None:
                return result
        return None

@contextmanager
def handle_exceptions(chain: ExceptionHandlerChain):
    try:
        yield
    except Exception as e:
        response = chain.handle(e)
        if response:
            return response
        raise

# Использование
exc_chain = ExceptionHandlerChain()

exc_chain.add(lambda e: {"status": 400} if isinstance(e, ValueError) else None)
exc_chain.add(lambda e: {"status": 404} if isinstance(e, KeyError) else None)
exc_chain.add(lambda e: {"status": 500} if isinstance(e, Exception) else None)

with handle_exceptions(exc_chain) as response:
    raise ValueError("Invalid input")
# response = {"status": 400}

#Python в стандартной библиотеке

#logging.Handler

import logging

# Chain of Responsibility в logging
logger = logging.getLogger("myapp")

# Обработчики образуют цепочку
console_handler = logging.StreamHandler()
file_handler = logging.FileHandler("app.log")

logger.addHandler(console_handler)
logger.addHandler(file_handler)

# Запрос (лог) проходит через все обработчики
logger.info("Message")

#exceptiongroup (Python 3.11+)

# Обработка группы исключений
def handle_exception_group(exc_group):
    for exc in exc_group.exceptions:
        if isinstance(exc, ValueError):
            print(f"ValueError: {exc}")
        elif isinstance(exc, TypeError):
            print(f"TypeError: {exc}")
        else:
            raise exc

#Что дальше?

Изучите тему Command и Undo/Redo для инкапсуляции действий как объектов.