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>
193 lines
6.3 KiB
Python
193 lines
6.3 KiB
Python
"""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})"
|