Skip to content
/ errata Public

Exception discovery and semantic boundary enforcement for Python. Prevent leaky abstractions. Make exception contracts explicit.

Notifications You must be signed in to change notification settings

getml/errata

Folders and files

NameName
Last commit message
Last commit date

Latest commit

 

History

43 Commits
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 

Repository files navigation

Errata

Exception discovery and semantic boundary enforcement for Python. Prevent leaky abstractions. Make exception contracts explicit.

Errata provides two complementary tools:

  1. Exception discovery — Static analysis for understanding what exceptions any Python package can raise
  2. The @boundary decorator — Semantic exception transformation at architectural boundaries

The Problem

Python's exception model makes it dangerously easy to leak implementation details:

# ✘ WRONG: SQLAlchemy exceptions leak to API consumers
def create_user(email: str, name: str) -> User:
    user = User(email=email, name=name)
    db.session.add(user)
    db.session.commit()  # May raise sqlalchemy.exc.IntegrityError
    return user

# Consumers now coupled to SQLAlchemy
try:
    user = create_user("test@example.com", "Test")
except sqlalchemy.exc.IntegrityError:  # Implementation detail leaked!
    print("User already exists")

The consequence: Changing databases breaks every consumer. Switch from SQLAlchemy to Django ORM? Every consumer must update their exception handling.

The Solution: Exception Boundaries

Mark semantic boundaries where exceptions transform from one layer to another:

from errata import boundary, on

# ✓ CORRECT: Explicit boundary prevents leaks
@boundary(
    on(sqlalchemy.exc.IntegrityError).raises(UserExistsError, "User already exists"),
    on(sqlalchemy.exc.OperationalError).raises(DatabaseError, "Database operation failed"),
)
def create_user(email: str, name: str) -> User:
    """Create user (BOUNDARY: database → domain).

    Raises:
        UserExistsError: User with this email exists.
        DatabaseError: Database operation failed.
    """
    user = User(email=email, name=name)
    db.session.add(user)
    db.session.commit()
    return user

# Consumers only see domain exceptions
try:
    user = create_user("test@example.com", "Test")
except UserExistsError:  # Clean domain exception
    print("User already exists")

The benefit: Switch databases freely. SQLAlchemy exceptions never escape the boundary. Your API contract stays stable.

Installation

uv pip install errata  # or: pip install errata

Quick Start

1. Define Domain Exceptions

class ConfigError(Exception):
    """Configuration error."""
    def __init__(self, message: str, *, path: str | None = None):
        super().__init__(message)
        self.path = path

2. Mark Boundaries

from errata import boundary, on

@boundary(
    on(FileNotFoundError, PermissionError).raises(ConfigError, "Cannot read config file"),
    on(json.JSONDecodeError).raises(ConfigError, "Invalid JSON in config"),
    on(ValueError).raises(ConfigError, "Invalid configuration"),
)
def load_config(path: str) -> dict:
    """Load configuration (BOUNDARY: file system → application).

    Raises:
        ConfigError: Configuration cannot be loaded or is invalid.
    """
    with open(path) as f:
        data = json.load(f)
    _validate_config(data)
    return data

3. Use Normally

try:
    config = load_config("config.json")
except ConfigError as e:
    print(f"Config error: {e}")
    print(f"Root cause: {e.__cause__}")  # Original exception preserved

The Two Rules

The @boundary decorator implements The Exception Locality Principle:

Exceptions should be handled at the level where they become meaningful, and transformed at boundaries where their meaning changes.

This gives us two simple rules:

Rule 1: Exceptions Flow to Their Natural Boundary

Internal code (within a semantic layer):

  • Let exceptions propagate by default
  • Document what flows through in docstrings
  • Only catch when recovery is possible at this level
def _read_file(path: str) -> str:
    """Read file contents.

    Raises:
        OSError: File cannot be read.
    """
    # Let OSError propagate - no try/except
    with open(path, encoding="utf-8") as f:
        return f.read()

def _parse_json(content: str) -> dict:
    """Parse JSON content.

    Raises:
        json.JSONDecodeError: Invalid JSON.
    """
    # Let JSONDecodeError propagate - no try/except
    return json.loads(content)

Rule 2: Boundaries Transform Exceptions

At semantic boundaries (between layers):

  • Use @boundary decorator to mark the boundary
  • Map lower-level exceptions to current layer's domain
  • Preserve cause chain for debugging
@boundary(
    on(FileNotFoundError, PermissionError).raises(ConfigError, "Cannot read config file"),
    on(json.JSONDecodeError).raises(ConfigError, "Invalid JSON in config"),
    on(ValueError).raises(ConfigError, "Invalid configuration"),
)
def load_config(path: str) -> dict:
    """Load configuration from JSON file (BOUNDARY: file system → application).

    Raises:
        ConfigError: Configuration cannot be loaded or is invalid.
    """
    content = _read_file(path)
    data = _parse_json(content)
    _validate_config(data)
    return data

Core Features

Multiple Source Exceptions

Map several exceptions to a single domain error:

@boundary(
    on(ConnectionError, TimeoutError, SSLError).raises(NetworkError, "Network failure"),
)
def fetch_data(url: str) -> dict:
    ...

Message Factories

Extract context from vendor exceptions:

@boundary(
    on(KeyError).raises(
        ConfigError,
        lambda e: f"Missing required key: {e.args[0]}"
    ),
)
def get_config_value(key: str, data: dict) -> Any:
    return data[key]

Guard Predicates

Branch by exception properties:

@boundary(
    on(requests.HTTPError).raises(
        NotFoundError,
        "Resource not found",
        when=lambda e: e.response.status_code == 404
    ),
    on(requests.HTTPError).raises(
        APIError,
        "API error",
        when=lambda e: e.response.status_code >= 500
    ),
)
def fetch_user(user_id: str) -> User:
    resp = requests.get(f"/users/{user_id}")
    resp.raise_for_status()
    return User(**resp.json())

Context Extraction

Copy structured data from vendor exceptions:

@boundary(
    on(sqlalchemy.exc.IntegrityError).raises(
        UserExistsError,
        "User already exists",
        context=lambda e: {"constraint": e.orig.diag.constraint_name}
    ),
)
def create_user(email: str) -> User:
    ...

# Usage: access extracted context
try:
    user = create_user("test@example.com")
except UserExistsError as e:
    print(e.constraint)  # "users_email_key"

Async Support

Works seamlessly with async functions:

@boundary(
    on(aiohttp.ClientError).raises(APIError, "API call failed"),
)
async def fetch_data(url: str) -> dict:
    async with aiohttp.ClientSession() as session:
        async with session.get(url) as resp:
            return await resp.json()

Fallback for Unexpected Exceptions

Catch unmapped exceptions at outermost boundaries:

@boundary(
    on(ValueError).raises(ValidationError, "Validation failed"),
    fallback=ServiceError,  # Wrap ANY unmapped exception
)
def handle_request(request: Request) -> Response:
    ...

Guidance: Use fallback only at your outermost boundaries. Internal functions should let bugs propagate.

Exception Discovery

Errata's static analysis discovers exceptions in any Python package:

errata discover rich
                                   Exceptions in 'rich'
┏━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ Exceptions                 ┃ Description                                                  ┃
┡━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
│ ColorParseError            │ The color could not be parsed.                               │
│ CaptureError               │ An error in the Capture context manager.                     │
│ ConsoleError               │ An error in console operation.                               │
│ ├── LiveError              │ Error related to Live display.                               │
│ ├── MarkupError            │ Markup was badly formatted.                                  │
│ └── StyleSyntaxError       │ Style was badly formatted.                                   │
└────────────────────────────┴──────────────────────────────────────────────────────────────┘

Show Usage Locations

errata discover --show-locations json
                                                                        Exceptions in 'json'
┏━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┳━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┓
┃ Exceptions      ┃ Description                                      ┃ Locations                         ┃
┡━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━╇━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━━┩
│ JSONDecodeError │ Subclass of ValueError with additional props... │ json (L335)                       │
│                 │                                                  │ json.decoder (L68, L86, L100...)  │
│                 │                                                  │ json.decoder.JSONDecoder (L348)   │
└─────────────────┴──────────────────────────────────────────────────┴───────────────────────────────────┘

Python API

from errata import find_exceptions, build_exception_hierarchy

# Find all exceptions in a package
exceptions = find_exceptions("mypackage")

# Build complete hierarchy with usage information
hierarchy = build_exception_hierarchy("mypackage", collect_usage=True)

# Display results
for exc in hierarchy.get_all_exceptions():
    print(f"{exc.qualified_name}: {len(exc.usage_locations)} usages")

Real-World Examples

Database Layer

@boundary(
    on(sqlalchemy.exc.IntegrityError).raises(UserExistsError, "User already exists"),
    on(sqlalchemy.exc.OperationalError).raises(DatabaseError, "Database operation failed"),
    on(sqlalchemy.orm.exc.NoResultFound).raises(UserNotFoundError, "User not found"),
)
def create_user(email: str, name: str) -> User:
    """Create user (BOUNDARY: database → domain).

    Raises:
        UserExistsError: User with this email exists.
        DatabaseError: Database operation failed.
    """
    user = User(email=email, name=name)
    db.session.add(user)
    db.session.commit()
    return user

HTTP Client

@boundary(
    on(requests.Timeout).raises(APITimeoutError, "Request timed out"),
    on(requests.HTTPError).raises(APIError, "HTTP request failed"),
    on(requests.ConnectionError).raises(APIError, "Connection failed"),
)
def fetch_user_data(user_id: str) -> dict:
    """Fetch user data from external API (BOUNDARY: HTTP → application).

    Raises:
        APITimeoutError: Request timed out.
        APIError: API request failed.
    """
    response = requests.get(f"https://api.example.com/users/{user_id}")
    response.raise_for_status()
    return response.json()

FastAPI Integration

from fastapi import FastAPI, HTTPException
from errata import boundary, on

app = FastAPI()

# Domain layer with boundary
@boundary(
    on(sqlalchemy.exc.IntegrityError).raises(DatabaseError, "Record exists"),
    on(requests.HTTPError).raises(APIError, "API failed"),
)
def create_user_operation(email: str) -> User:
    """Domain logic with sealed abstraction boundary."""
    user = User(email=email)
    db.session.add(user)
    db.session.commit()
    return user

# Framework layer maps domain exceptions to HTTP responses
@app.exception_handler(DatabaseError)
async def handle_db_error(request, exc: DatabaseError):
    return JSONResponse(
        status_code=409,
        content={"error": str(exc), "type": "database_error"}
    )

@app.post("/users")
async def create_user_endpoint(email: str):
    user = create_user_operation(email)  # Raises DatabaseError
    return {"id": user.id}

Design Patterns

Exception Hierarchies

Create hierarchies that allow catching at different levels:

# Base for entire domain
class OrderException(Exception):
    """Base for all order-related errors."""
    pass

# Group by operation type
class CreateTaskFailed(OrderException):
    """Task creation failed."""
    pass

class StoreTaskFailed(OrderException):
    """Task storage failed."""
    pass

# Specific operations
class CreatePickUpTaskFailed(CreateTaskFailed):
    """Failed to create pick-up task."""
    pass

Storing Context

Use keyword-only parameters for context while remaining compatible with @boundary:

class ConfigError(Exception):
    """Configuration error."""
    def __init__(self, message: str, *, path: str | None = None):
        super().__init__(message)
        self.path = path

# @boundary can still instantiate: ConfigError(message)
# You can add context later:
try:
    config = load_config(path)
except ConfigError as e:
    e.path = path  # Add context after boundary
    raise

When to Use Errata

Exception Discovery

Valuable for any Python project:

  • Understanding third-party library error contracts
  • Auditing exception usage in large codebases
  • Generating documentation for API boundaries
  • No adoption costs—use it as a standalone tool

The @boundary Decorator

Best suited for:

  • Preventing leaky abstractions (vendor exceptions escaping)
  • Creating stable API contracts across architectural layers
  • Teams integrating many third-party libraries
  • Projects where switching vendors should not break consumers

Not needed for:

  • Small scripts or prototypes
  • Internal utilities with no API consumers
  • Simple applications with few dependencies

Key Benefits

  • Prevents leaky abstractions — Vendor exceptions never escape boundaries
  • Explicit contracts — Exception mappings visible at function definition
  • Superior debugging — Full tracebacks preserved via __cause__ chain
  • Async support — Works with both sync and async functions
  • Type-safe — Type checkers understand the transformation
  • Testable — Can test boundaries in isolation
  • Framework-agnostic — Works with FastAPI, Flask, Django, CLI, etc.

Testing Boundaries

Test both transformation and cause preservation:

def test_boundary_transforms_exceptions():
    with pytest.raises(ConfigError, match="Config not found"):
        load_config("nonexistent.json")

def test_boundary_preserves_cause_chain():
    with pytest.raises(ConfigError) as exc_info:
        load_config("nonexistent.json")

    # Cause chain preserved for debugging
    assert isinstance(exc_info.value.__cause__, FileNotFoundError)

Architecture

errata/
├── primitives/          # Core boundary primitives
│   ├── boundary.py      # @boundary decorator
│   └── mappings.py      # Exception mapping types (on, ExceptionMap)
├── discovery/           # Static analysis tools
│   ├── static.py        # AST-based exception discovery
│   └── hierarchy.py     # Exception hierarchy analysis
└── cli/                 # Command-line interface
    └── main.py          # CLI application

Development

git clone https://github.com/getml/errata.git
cd errata
mise install
uv sync
pytest

Related Reading


Errata helps you move from leaky, implicit exception contracts to explicit, semantic boundaries that prevent vendor lock-in and make your APIs stable.

About

Exception discovery and semantic boundary enforcement for Python. Prevent leaky abstractions. Make exception contracts explicit.

Resources

Stars

Watchers

Forks

Releases

No releases published

Packages

No packages published

Languages