"""Error handling contracts for services. Defines the three allowed error handling patterns so callers know what to expect from any service method. Pattern Selection ================ - ABORT_ON_ERROR: Operations where failure must propagate (auth, writes, config changes) - RETURN_DEFAULT: Informational reads where partial data is acceptable - PARTIAL_RESULT: Operations on collections where some items may fail independently Switching patterns is a breaking change — document in changelog. Type Annotations =============== Each pattern has a corresponding type to make contracts machine-checkable: | Pattern | Type | Use when | |-----------------|--------------------------------|-----------------------------------------| | ABORT_ON_ERROR | ``T`` (plain return) | Raises ``DomainError`` on failure | | RETURN_DEFAULT | ``T | None`` | Returns ``None`` on failure | | PARTIAL_RESULT | ``PartialResult[T]`` | Returns ``(T, list[ErrorEntry])`` | Example (ABORT_ON_ERROR — raises on failure):: async def start_jail(socket_path: str, name: str) -> None: '''Start a stopped fail2ban jail. Error contract: ABORT_ON_ERROR. Raises JailNotFoundError (404), JailOperationError (409), Fail2BanConnectionError (503). ''' ... Example (RETURN_DEFAULT — returns None on failure):: async def get_settings(socket_path: str) -> DomainServerSettingsResult | None: '''Return current fail2ban server-level settings. Error contract: RETURN_DEFAULT. Returns None if socket is unreachable. ''' ... Example (PARTIAL_RESULT — returns (result, errors) on failure):: async def fetch_all_bans( socket_path: str, jail_names: list[str] ) -> PartialResult[DomainJailBannedIps]: '''Fetch bans from multiple jails. Error contract: PARTIAL_RESULT. Returns (results, errors) where errors contains one entry per jail that failed. ''' ... """ from __future__ import annotations import enum from typing import Self, TypeVar ABORT_ON_ERROR = "abort_on_error" """Raise an exception. Router converts to HTTP. Use for auth, writes, state changes.""" RETURN_DEFAULT = "return_default" """Return empty result and log warning. Never raises. Use for informational reads.""" PARTIAL_RESULT = "partial_result" """Return (result, errors) tuple. Use for batch operations on collections.""" T = TypeVar("T") class ErrorContract(enum.Enum): """Machine-checkable error handling pattern for a service or method. Use the ``value`` attribute to get the string constant for docstrings and comparisons. Use ``from_value()`` to look up by string. Example:: contract = ErrorContract.from_value(ABORT_ON_ERROR) if contract == ErrorContract.ABORT_ON_ERROR: ... """ ABORT_ON_ERROR = ABORT_ON_ERROR RETURN_DEFAULT = RETURN_DEFAULT PARTIAL_RESULT = PARTIAL_RESULT @classmethod def from_value(cls, value: str) -> Self: """Return the enum member matching *value*. Raises: ValueError: If *value* is not one of the three patterns. """ for member in cls: if member.value == value: return member raise ValueError(f"Unknown error contract {value!r}") class ServiceErrorContract: """Documents the error handling pattern for a service or method. Callers use this to understand how errors affect the return value: ABORT_ON_ERROR Raise an exception. Router handles it, converts to HTTP response. Used for: authentication, authorization, write operations, state changes, and any operation where partial success is meaningless. RETURN_DEFAULT Return empty/None result and log a warning. Caller gets a valid result with no items, not an error. Used for: informational reads (list, get) where infrastructure unavailability should not block the UI. PARTIAL_RESULT Return a result that contains both successful items and a list of errors. Caller decides what to do with each. Used for: batch operations, multi-item fetches where one item failing does not invalidate the rest. """ ABORT_ON_ERROR = ABORT_ON_ERROR RETURN_DEFAULT = RETURN_DEFAULT PARTIAL_RESULT = PARTIAL_RESULT @classmethod def doc(cls, pattern: str, *, since: str | None = None) -> str: """Return a docstring fragment describing the error pattern.""" desc = { ABORT_ON_ERROR: "Raises exceptions on error. Router handles conversion to HTTP.", RETURN_DEFAULT: "Returns empty result and logs warning on error. Never raises.", PARTIAL_RESULT: "Returns (result, errors) tuple. Errors collected, not raised.", }[pattern] if since: return f"{desc} (Since: {since})" return desc # --------------------------------------------------------------------------- # Typed patterns (incremental migration) # --------------------------------------------------------------------------- class ErrorEntry: """Single error from a PARTIAL_RESULT operation. Attributes: context: Human-readable description of what failed. cause: The underlying exception, if available. """ __slots__ = ("context", "cause") def __init__(self, context: str, *, cause: Exception | None = None) -> None: self.context = context self.cause = cause def __repr__(self) -> str: cls = type(self) return f"{cls.__name__}(context={self.context!r}, cause={self.cause!r})" class PartialResult: """Result of a PARTIAL_RESULT operation. Attributes: ok: The successful result (may be empty list/dict). errors: Errors encountered during the operation. Usage:: result: PartialResult[DomainJailBannedIps] log.warning("some_jails_failed", error_count=len(result.errors)) """ __slots__ = ("ok", "errors") def __init__(self, ok: T, errors: list[ErrorEntry]) -> None: self.ok: T = ok self.errors: list[ErrorEntry] = errors def __repr__(self) -> str: cls = type(self) return f"{cls.__name__}(ok={self.ok!r}, errors={self.errors!r})"