Metadata-Version: 2.4
Name: domain-errors
Version: 0.2.0
Summary: Typed domain error hierarchy with wrapping and chaining for Python services
Project-URL: Homepage, https://pypi.org/project/domain-errors/
Project-URL: Repository, https://github.com/jekhator/domain-errors
Project-URL: Issues, https://github.com/jekhator/domain-errors/issues
Project-URL: Changelog, https://github.com/jekhator/domain-errors/blob/main/CHANGELOG.md
Author: James Ekhator
License: Apache-2.0
License-File: LICENSE
Keywords: chaining,error-handling,errors,exceptions,typed
Classifier: Development Status :: 3 - Alpha
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: Apache Software License
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.11
Classifier: Programming Language :: Python :: 3.12
Classifier: Typing :: Typed
Requires-Python: >=3.11
Description-Content-Type: text/markdown

# domain-errors

[![PyPI](https://img.shields.io/pypi/v/domain-errors)](https://pypi.org/project/domain-errors/)
[![CI](https://github.com/jekhator/domain-errors/actions/workflows/ci.yml/badge.svg)](https://github.com/jekhator/domain-errors/actions)
[![License](https://img.shields.io/badge/license-Apache%202.0-blue)](LICENSE)
[![Python](https://img.shields.io/badge/python-3.11%2B-blue)](https://www.python.org/downloads/)

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.

```python
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

```bash
pip install domain-errors
```

or with uv:

```bash
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:

```python
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:

```python
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:

```python
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:

```python
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:

```python
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:**
```python
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.

```python
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.

```python
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:

```python
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:

```python
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:**

```python
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):**

```python
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:

```python
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](docs/apps/domain_error.md): class contract, subclassing, message/context semantics
- **ErrorChain:** [docs/apps/chain.md](docs/apps/chain.md): wrap, history, crossings, logging integration
- **@wrap_errors Decorator:** [docs/apps/wrap_errors.md](docs/apps/wrap_errors.md): declarative error wrapping for functions and async callables
- **Architecture:** [docs/apps/diagrams.md](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](https://github.com/jekhator).
