Exception discovery and semantic boundary enforcement for Python. Prevent leaky abstractions. Make exception contracts explicit.
Errata provides two complementary tools:
- Exception discovery — Static analysis for understanding what exceptions any Python package can raise
- The @boundary decorator — Semantic exception transformation at architectural boundaries
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.
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.
uv pip install errata # or: pip install errataclass ConfigError(Exception):
"""Configuration error."""
def __init__(self, message: str, *, path: str | None = None):
super().__init__(message)
self.path = pathfrom 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 datatry:
config = load_config("config.json")
except ConfigError as e:
print(f"Config error: {e}")
print(f"Root cause: {e.__cause__}") # Original exception preservedThe @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:
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)At semantic boundaries (between layers):
- Use
@boundarydecorator 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 dataMap several exceptions to a single domain error:
@boundary(
on(ConnectionError, TimeoutError, SSLError).raises(NetworkError, "Network failure"),
)
def fetch_data(url: str) -> dict:
...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]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())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"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()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.
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. │
└────────────────────────────┴──────────────────────────────────────────────────────────────┘❯ 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) │
└─────────────────┴──────────────────────────────────────────────────┴───────────────────────────────────┘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")@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@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()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}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."""
passUse 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
raiseValuable 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
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
- 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.
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)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
git clone https://github.com/getml/errata.git
cd errata
mise install
uv sync
pytest- Proposal: The Two Rules Approach — Full design rationale
- Miguel Grinberg: Error Handling in Python
- How to Structure Exceptions in Python Like a Pro
Errata helps you move from leaky, implicit exception contracts to explicit, semantic boundaries that prevent vendor lock-in and make your APIs stable.