feat(services): add ErrorContract enum and PartialResult type

Add typed wrappers for error handling patterns in error_handling.py:

- ErrorContract(enum): machine-checkable pattern selector with
  from_value() helper and string constants matching the existing
  ABORT_ON_ERROR/RETURN_DEFAULT/PARTIAL_RESULT module-level values
- ErrorEntry: typed error container for PARTIAL_RESULT (context + cause)
- PartialResult[T]: typed result wrapper for PARTIAL_RESULT operations

Existing string constants preserved for backward compat.
Updated module docstring with type annotation table and examples.

Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
2026-05-03 22:46:47 +02:00
parent a2afec2d1e
commit edebf1a339

View File

@@ -10,10 +10,56 @@ Pattern Selection
- 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."""
@@ -23,6 +69,39 @@ RETURN_DEFAULT = "return_default"
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.
@@ -62,3 +141,52 @@ class ServiceErrorContract:
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})"