domain-errors 0.2.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. Works on both functions and classes (fans out to all public methods, preserving sync/async dispatch).

Sync and async functions:

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()

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

Service classes (class-level wrapping):

from domain_errors import DomainError, wrap_errors

class PaymentError(DomainError):
    code = "payment_error"
    domain = "payment"
    http_status = 402

@wrap_errors(PaymentError, catch=(ValueError,))
class PaymentService:
    def validate_amount(self, amount: int) -> bool:
        if amount <= 0:
            raise ValueError("Amount must be positive")
        return True

    async def process_payment(self, card: str, amount: int) -> str:
        if not card:
            raise ValueError("Card required")
        return f"Processed {amount}"

service = PaymentService()

assert service.validate_amount(100)

try:
    service.validate_amount(-50)
except PaymentError as err:
    print(f"Error: {err.message}")
    print(f"Context: {err.context}")
    print(f"Cause: {err.__cause__}")

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

Service Class Example

A complete service class pattern with error wrapping and context capture:

from domain_errors import DomainError, ErrorChain

class PaymentService:
    @staticmethod
    def process_payment(amount_cents: int) -> dict:
        try:
            if amount_cents <= 0:
                raise ValueError("Amount must be positive")
            if amount_cents > 999999:
                raise ValueError("Amount exceeds maximum")
            return {"status": "success", "amount_cents": amount_cents}
        except ValueError as e:
            raise ErrorChain.wrap(
                e,
                as_=PaymentError,
                message=f"Payment validation failed: {e}",
                amount_cents=amount_cents
            ) from e

class PaymentError(DomainError):
    code = "payment_error"
    domain = "payment"
    http_status = 400
    default_message = "Payment processing failed."

# Usage
try:
    result = PaymentService.process_payment(5000)
    print(f"Success: {result}")
except PaymentError as err:
    print(f"Error {err.code} in {err.domain}: {err.message}")
    print(f"Context: {err.context}")

Output:

✓ Successful payment: {'status': 'success', 'amount_cents': 5000}
✓ Caught PaymentError: code=payment_error, domain=payment
  Message: Payment validation failed: Amount must be positive
  Context: {'amount_cents': -100}
  Chain length: 2
✓ Caught PaymentError: amount_cents=9999999

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.