110 lines
4.2 KiB
Python
110 lines
4.2 KiB
Python
import json
|
|
from typing import Any, Dict, Callable, List
|
|
|
|
class MiddlewareError(Exception):
|
|
"""Исключение при работе middleware."""
|
|
|
|
|
|
class HumanInTheLoop:
|
|
"""
|
|
Класс реализует цепочку обработки данных с возможностью вмешательства человека.
|
|
"""
|
|
|
|
def __init__(self) -> None:
|
|
self._middleware: List[Callable[[Dict[str, Any]], Dict[str, Any]]] = []
|
|
self._handlers: Dict[str, Callable[[Dict[str, Any]], None]] = {}
|
|
|
|
def add_middleware(self, func: Callable[[Dict[str, Any]], Dict[str, Any]]) -> None:
|
|
"""
|
|
Добавляет функцию middleware в цепочку.
|
|
"""
|
|
if not callable(func):
|
|
raise MiddlewareError("Middleware должен быть вызываемым.")
|
|
self._middleware.append(func)
|
|
|
|
def register_handler(self, name: str, handler: Callable[[Dict[str, Any]], None]) -> None:
|
|
"""
|
|
Регистрирует обработчик с именем `name`.
|
|
"""
|
|
if not callable(handler):
|
|
raise MiddlewareError("Handler должен быть вызываемым.")
|
|
self._handlers[name] = handler
|
|
|
|
def _process(self, data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Последовательно применяет все middleware к данным.
|
|
"""
|
|
for mw in self._middleware:
|
|
try:
|
|
data = mw(data)
|
|
if not isinstance(data, dict):
|
|
raise MiddlewareError("Middleware должен вернуть словарь.")
|
|
except Exception as exc:
|
|
raise MiddlewareError(f"Ошибка в middleware {mw.__name__}: {exc}") from exc
|
|
return data
|
|
|
|
def run(self, raw_input: str) -> None:
|
|
"""
|
|
Точка входа. Принимает строку JSON, преобразует её,
|
|
запускает цепочку middleware и вызывает соответствующий обработчик.
|
|
"""
|
|
try:
|
|
payload = json.loads(raw_input)
|
|
except json.JSONDecodeError as exc:
|
|
raise MiddlewareError("Неверный формат JSON") from exc
|
|
|
|
processed = self._process(payload)
|
|
|
|
handler_name = processed.get("handler")
|
|
if not handler_name or handler_name not in self._handlers:
|
|
raise MiddlewareError(f"Неизвестный обработчик: {handler_name}")
|
|
|
|
try:
|
|
self._handlers[handler_name](processed)
|
|
except Exception as exc:
|
|
raise MiddlewareError(f"Ошибка в обработчике {handler_name}: {exc}") from exc
|
|
|
|
# ----------------- Пример middleware и обработчиков -----------------
|
|
|
|
def validate_mw(data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Проверяет наличие обязательных полей."""
|
|
if "payload" not in data:
|
|
raise MiddlewareError("Отсутствует поле 'payload'")
|
|
return data
|
|
|
|
def enrich_mw(data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""Добавляет метку времени."""
|
|
import datetime
|
|
data["timestamp"] = datetime.datetime.utcnow().isoformat()
|
|
return data
|
|
|
|
def echo_handler(data: Dict[str, Any]) -> None:
|
|
print("Echo:", data.get("payload"))
|
|
|
|
def sum_handler(data: Dict[str, Any]) -> None:
|
|
numbers = data.get("payload", [])
|
|
if not isinstance(numbers, list):
|
|
raise MiddlewareError("'payload' должен быть списком")
|
|
result = sum(numbers)
|
|
print(f"Sum: {result}")
|
|
|
|
# ----------------- Инициализация -----------------
|
|
if __name__ == "__main__":
|
|
hltl = HumanInTheLoop()
|
|
hltl.add_middleware(validate_mw)
|
|
hltl.add_middleware(enrich_mw)
|
|
|
|
hltl.register_handler("echo", echo_handler)
|
|
hltl.register_handler("sum", sum_handler)
|
|
|
|
# Пример входных данных
|
|
test_inputs = [
|
|
json.dumps({"handler": "echo", "payload": "Hello, world!"}),
|
|
json.dumps({"handler": "sum", "payload": [1, 2, 3, 4]}),
|
|
]
|
|
|
|
for inp in test_inputs:
|
|
try:
|
|
hltl.run(inp)
|
|
except MiddlewareError as e:
|
|
print("Ошибка:", e) |