domain-errors 0.1.0

Typed domain error hierarchy with wrapping and chaining for Python services

domain-errors

Typed domain error hierarchy with wrapping and chaining for Python services.

Define per-domain exception types with a structured contract (code, domain, HTTP status, retryability). Wrap foreign exceptions into domain errors, walk causation chains, and detect when errors cross domain boundaries—all ready for structured logging.

Why?

Service errors should carry semantic meaning: what failed (code), where it failed (domain), how to handle it (HTTP status, retryability), and why it happened (context). Cross-service calls risk mixing domains—database errors become API errors become payment errors. domain-errors makes this hierarchy explicit and traceable.

from domain_errors import DomainError, ErrorChain

class DatabaseError(DomainError):
    domain = "database"
    code = "db_error"
    http_status = 503
    retryable = True

try:
    user = db.query("SELECT * FROM users WHERE id = $1", user_id)
except Exception as e:
    raise ErrorChain.wrap(
        e,
        as_=DatabaseError,
        message=f"Failed to fetch user {user_id}",
        user_id=user_id
    ) from e

Installation

pip install domain-errors

or with uv:

uv add domain-errors

Requires Python 3.11+.

Quick Start

1. Define a domain error

Subclass DomainError with a code, domain, HTTP status, and message:

from domain_errors import DomainError

class PaymentError(DomainError):
    code = "payment_declined"
    domain = "payment"
    http_status = 402
    retryable = False
    default_message = "Payment was declined."

2. Raise with structured context

Include context keyword arguments for metrics and audit trails:

raise PaymentError(
    message="Card declined: insufficient funds",
    transaction_id="txn_xyz789",
    amount_cents=5000,
    currency="USD"
)

3. Wrap foreign exceptions

Convert caught exceptions into domain errors, preserving the chain:

try:
    result = stripe.Charge.create(amount=5000, source=card_token)
except stripe.error.CardError as e:
    raise ErrorChain.wrap(
        e,
        as_=PaymentError,
        message=f"Card charge failed: {e.user_message}",
        transaction_id=e.http_body["id"]
    ) from e

4. Trace and log the chain

Walk the exception chain to find domain crossings and log structured data:

try:
    process_payment(amount_cents=5000)
except DomainError as err:
    # Get the full chain
    links = ErrorChain.history(err)
    for link in links:
        logger.error("Error in chain", extra=link.to_log_extra())
    
    # Find where errors crossed domains
    crossings = ErrorChain.crossings(err)
    for crossing in crossings:
        logger.warning("Domain boundary crossed", extra=crossing.to_log_extra())
    
    # Return API response
    return {
        "status": err.http_status,
        "code": err.code,
        "message": err.message
    }

Public API

DomainError

Base class for domain-specific exceptions. Define subclasses with a contract:

class MyError(DomainError):
    code: str = "my_error"           # Unique error code
    domain: str = "my_domain"        # Namespace (auth, database, etc.)
    http_status: int = 400           # HTTP status for API responses
    retryable: bool = False          # Transient failure?
    default_message: str = "..."     # Fallback message

Constructor:

error = MyError(
    message="Custom message",  # optional; defaults to default_message
    **context                  # structured data: user_id=123, txn_id="x"
)

Attributes:

• message: str — the error message
• context: dict[str, object] — structured context data

ErrorChain

Static methods for wrapping, walking, and analyzing exception chains.

Methods:

• ErrorChain.wrap(err, as_=ErrorType, message=None, **context) — Construct a typed domain error for the caller to raise from err.
• ErrorChain.history(err, classifiers=()) — Walk the exception chain; return a tuple of ChainLink objects.
• ErrorChain.crossings(err, classifiers=()) — Find causation hops where errors crossed domains; return a tuple of DomainCrossing objects.

ChainLink

One hop in an exception chain, ready for structured logging.

link = ChainLink(
    type_name="TimeoutError",
    message="[Errno 110] Connection timed out",
    code=None,
    domain="python",
    via=ChainVia.CONTEXT,
    context={}
)

# Convert to JSON-ready dict for logger extra:
logger.error("Exception", extra=link.to_log_extra())

DomainCrossing

A causation hop where errors changed domains.

crossing = DomainCrossing(
    cause=ChainLink(...),  # e.g., TimeoutError (python)
    effect=ChainLink(...)  # e.g., APIError (api)
)

logger.warning("Domain boundary", extra=crossing.to_log_extra())

DomainClassifier (Protocol)

Optional contract for resolving domains of foreign exceptions:

class MyClassifier:
    def classify(self, err: BaseException) -> str | None:
        """Return domain name, or None if not applicable."""
        if isinstance(err, TimeoutError):
            return "stdlib"
        return None

links = ErrorChain.history(err, classifiers=(MyClassifier(),))

Built-in classifiers are provided for standard library, HTTP-client, and AWS SDK exceptions:

from domain_errors.domains.python import python
from domain_errors.domains.http import http
from domain_errors.domains.cloud import cloud

# Classify stdlib exceptions
links = ErrorChain.history(err, classifiers=(python,))

# Classify httpx, requests, and aiohttp exceptions
links = ErrorChain.history(err, classifiers=(http,))

# Classify botocore and boto3 exceptions
links = ErrorChain.history(err, classifiers=(cloud,))

# Compose multiple classifiers
links = ErrorChain.history(err, classifiers=(python, http, cloud))

@wrap_errors

Decorator for automatic error wrapping. Catches specified exceptions, wraps them into a target DomainError, and captures function arguments for structured logging:

from domain_errors import DomainError, wrap_errors

class StorageError(DomainError):
    code = "storage_error"
    domain = "storage"
    http_status = 503

@wrap_errors(StorageError, catch=(FileNotFoundError, IOError))
def read_config(path: str) -> str:
    with open(path) as f:
        return f.read()

# Manual wrapping (before)
try:
    config = read_config("/etc/app.yaml")
except FileNotFoundError as e:
    raise ErrorChain.wrap(e, as_=StorageError, path=path) from e

# Declarative wrapping (after)
@wrap_errors(StorageError, catch=(FileNotFoundError, IOError))
def read_config(path: str) -> str:
    with open(path) as f:
        return f.read()

try:
    config = read_config("/etc/app.yaml")
except StorageError as err:
    # err.context includes {'path': '/etc/app.yaml'}
    # err.__cause__ is the original FileNotFoundError
    pass

DomainError instances pass through unchanged; works on sync and async callables; set capture=False to omit arguments (e.g., for secrets).

Documentation

• DomainError Base: docs/apps/domain_error.md — class contract, subclassing, message/context semantics
• ErrorChain: docs/apps/chain.md — wrap, history, crossings, logging integration
• @wrap_errors Decorator: docs/apps/wrap_errors.md — declarative error wrapping for functions and async callables
• Architecture: docs/apps/diagrams.md — data flow and domain relationships

License

Apache 2.0 — see LICENSE file.

Contributing

Contributions welcome via pull requests. This library is maintained by James Ekhator.