Цепочки обработчиков. Middleware-паттерн в веб-фреймворках. Обработка ошибок и short-circuit evaluation.
Цепочка превращает множество обработчиков в единый конвейер.
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. Трудно тестировать отдельные обработчики# ✅ ХОРОШО: Каждый обработчик — отдельный класс
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 — ни один обработчик не справился# ✅ Проще: цепочка через список
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 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: {...}# ✅ 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}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")# Обработка группы исключений
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| Паттерн | Когда использовать | Pythonic-реализация |
|---|---|---|
| Chain of Responsibility | Последовательная обработка, конвейер | Классы с _next или список функций |
| Middleware | HTTP-запросы, обработка исключений | Декораторы + композиция функций |
| Short-circuit | Раннее прерывание цепочки | Возврат результата до _handle_next |
Главный принцип: Каждый обработчик решает, обработать запрос или передать дальше.
Изучите тему Command и Undo/Redo для инкапсуляции действий как объектов.
Вопросы ещё не добавлены
Вопросы для этой подтемы ещё не добавлены.