Skip to content

Handle Domain Errors

DDD CQRS ES

A domain exception starts at an aggregate invariant and ends as an API response. Here is how to raise one, how it propagates, and where to handle it along the way.

Exception hierarchy

Protean provides domain-specific exceptions that propagate naturally through the application layers:

Exception When to use Typical HTTP mapping
ValidationError Invariant violation, field validation failure 400
InvalidDataError Data type or value mismatch 400
ObjectNotFoundError Aggregate not found in persistence 404
InvalidStateError Operation invalid for current aggregate state 409
InvalidOperationError Business rule prevents the operation 422
ExpectedVersionError Optimistic concurrency conflict (ES) 409

Raising errors in aggregates

From invariants

Invariants raise ValidationError when business rules are violated. The error dict uses "_entity" as the key for aggregate-level errors:

from protean.exceptions import ValidationError

@domain.aggregate
class Order:
    items = HasMany("OrderItem")
    status = String(default="draft")

    @invariant.post
    def must_have_items(self):
        if not self.items:
            raise ValidationError(
                {"_entity": ["Order must contain at least one item"]}
            )

From domain methods

Use InvalidStateError or InvalidOperationError for explicit business logic checks:

from protean.exceptions import InvalidStateError

@domain.aggregate
class Order:
    status = String(default="draft")

    def cancel(self):
        if self.status == "shipped":
            raise InvalidStateError("Cannot cancel a shipped order")
        self.status = "cancelled"

Error propagation through layers

Protean exceptions propagate naturally, don't catch them in handlers or services unless you need to translate them:

Aggregate method
  └── raises ValidationError / InvalidStateError
        └── propagates through Command Handler
              └── propagates through domain.process()
                    └── propagates through API endpoint
                          └── mapped to HTTP response by exception handler

In command handlers

Let exceptions propagate. Don't wrap them:

@domain.command_handler(part_of=Order)
class OrderCommandHandler:
    @handle(PlaceOrder)
    def place_order(self, command):
        order = Order(
            customer_id=command.customer_id,
            items=command.items,  # ValidationError if empty
        )
        order.place()  # InvalidStateError if invalid transition
        current_domain.repository_for(Order).add(order)

In application services

Same principle, let domain exceptions propagate to the caller:

@domain.application_service(part_of=Order)
class OrderService:
    @use_case
    def place_order(self, customer_id, items):
        order = Order(customer_id=customer_id, items=items)
        order.place()
        current_domain.repository_for(Order).add(order)
        return order.id

Repository errors

Method Raises When
repo.get(id) ObjectNotFoundError ID not found
repo.find_by(**kwargs) ObjectNotFoundError No match found
repo.find_by(**kwargs) TooManyObjectsError Multiple matches
repo.find(criteria) (never) Returns empty ResultSet
repo.exists(criteria) (never) Returns bool

Use repo.find() or repo.exists() when absence is expected, and repo.get() when absence is an error:

# Absence is an error -- let it raise
order = repo.get(order_id)  # ObjectNotFoundError → 404

# Absence is expected -- check gracefully
if repo.exists(Q(email=email)):
    raise InvalidOperationError("Email already registered")

Unit of Work errors

Transaction failures during UnitOfWork.commit() raise:

  • ExpectedVersionError: Optimistic concurrency conflict in event-sourced aggregates. The client should retry with fresh state.
  • TransactionError: Wraps the underlying database error. The original exception is available as exc.__cause__.

Mapping to HTTP responses

Use register_exception_handlers to automatically map domain exceptions to HTTP status codes in FastAPI:

from protean.integrations.fastapi import register_exception_handlers

app = FastAPI()
register_exception_handlers(app)
Exception HTTP Status Response body
ValidationError 400 {"error": {"field": ["message"]}}
InvalidDataError 400 {"error": {"field": ["message"]}}
ValueError 400 {"error": "message"}
ObjectNotFoundError 404 {"error": "message"}
InvalidStateError 409 {"error": "message"}
InvalidOperationError 422 {"error": "message"}

Your endpoints don't need try/except blocks:

@app.post("/orders", status_code=201)
async def place_order(payload: dict):
    # ValidationError → 400, ObjectNotFoundError → 404, etc.
    current_domain.process(PlaceOrder(**payload))
    return {"status": "accepted"}

See FastAPI Integration for full setup.

Testing error conditions

Use pytest.raises:

import pytest
from protean.exceptions import ValidationError

def test_order_requires_items():
    with pytest.raises(ValidationError) as exc:
        Order(customer_id="c1", items=[])
    assert "must contain at least one item" in str(exc.value)

Deprecated: assert_valid / assert_invalid

The protean.testing.assert_valid and assert_invalid helpers still exist but are deprecated and will be removed in v0.18.0. Prefer pytest.raises. Replace assert_invalid(lambda: op(), message="X") with with pytest.raises(ValidationError, match="X"): op(), and assert_valid(lambda: op()) with a plain call to op().

See also