File size: 2,525 Bytes
152be26 | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 | """Allowlisted handler registry for workflow task execution.
The registry is the application-controlled boundary between a workflow definition
and executable Python code. Workflow data may name a handler, but only handlers
explicitly registered here can ever be resolved for execution.
"""
from __future__ import annotations
from collections.abc import Callable
from typing import Any, TypeAlias
Handler: TypeAlias = Callable[[dict[str, Any]], dict[str, Any]]
class HandlerRegistryError(ValueError):
"""Base error for invalid handler-registry operations."""
class DuplicateHandlerError(HandlerRegistryError):
"""Raised when code tries to register the same handler identifier twice."""
class UnknownHandlerError(HandlerRegistryError):
"""Raised when a workflow requests a handler that is not allowlisted."""
class HandlerRegistry:
"""Explicit allowlist mapping stable handler IDs to Python callables."""
def __init__(self) -> None:
self._handlers: dict[str, Handler] = {}
def register(self, handler_id: str, handler: Handler) -> None:
"""Register one approved handler under a stable application-owned ID."""
normalized_id = handler_id.strip()
if not normalized_id:
raise HandlerRegistryError("handler_id must not be blank")
if not callable(handler):
raise HandlerRegistryError("handler must be callable")
if normalized_id in self._handlers:
raise DuplicateHandlerError(
f"handler '{normalized_id}' is already registered"
)
self._handlers[normalized_id] = handler
def resolve(self, handler_id: str) -> Handler:
"""Return an approved handler or reject an unknown identifier."""
normalized_id = handler_id.strip()
if not normalized_id:
raise UnknownHandlerError("handler_id must not be blank")
try:
return self._handlers[normalized_id]
except KeyError as exc:
raise UnknownHandlerError(
f"handler '{normalized_id}' is not registered"
) from exc
def contains(self, handler_id: str) -> bool:
"""Return whether a handler identifier is currently allowlisted."""
normalized_id = handler_id.strip()
return bool(normalized_id) and normalized_id in self._handlers
def registered_ids(self) -> tuple[str, ...]:
"""Return registered handler IDs in deterministic sorted order."""
return tuple(sorted(self._handlers))
|