from dataclasses import dataclass, field
from datetime import datetime, timezone
from decimal import Decimal
def _now():
return datetime.now(timezone.utc)
@dataclass(frozen=True)
class DomainEvent:
occurred_at: datetime = field(default_factory=_now)
@dataclass(frozen=True)
class OrderPlaced(DomainEvent):
order_id: str = ""
customer_email: str = ""
total: Decimal = Decimal("0")
line_items: tuple = ()
@dataclass(frozen=True)
class PaymentCaptured(DomainEvent):
order_id: str = ""
amount: Decimal = Decimal("0")
processor_ref: str = ""
import logging
from collections import defaultdict
from typing import Callable, Dict, List, Type
from events import DomainEvent
logger = logging.getLogger(__name__)
Handler = Callable[[DomainEvent], None]
class EventBus:
def __init__(self):
self._handlers: Dict[Type[DomainEvent], List[Handler]] = defaultdict(list)
def subscribe(self, event_type: Type[DomainEvent], handler: Handler) -> None:
self._handlers[event_type].append(handler)
def handles(self, event_type: Type[DomainEvent]):
def decorator(func: Handler) -> Handler:
self.subscribe(event_type, func)
return func
return decorator
def publish(self, event: DomainEvent) -> None:
handlers = self._handlers.get(type(event), [])
if not handlers:
logger.debug("No handlers for %s", type(event).__name__)
return
failures = self._dispatch(event, handlers)
if failures:
logger.warning("%d handler(s) failed for %s", len(failures), type(event).__name__)
def _dispatch(self, event: DomainEvent, handlers: List[Handler]) -> List[Exception]:
failures: List[Exception] = []
for handler in handlers:
try:
handler(event)
except Exception as exc: # isolate one bad subscriber from the rest
logger.exception("Handler %r raised on %s", handler, type(event).__name__)
failures.append(exc)
return failures
bus = EventBus()
from decimal import Decimal
from event_bus import bus
from events import OrderPlaced, PaymentCaptured
@bus.handles(OrderPlaced)
def send_confirmation_email(event: OrderPlaced) -> None:
print(f"Emailing {event.customer_email}: order {event.order_id} received")
@bus.handles(OrderPlaced)
def update_inventory(event: OrderPlaced) -> None:
for sku, qty in event.line_items:
print(f"Reserving {qty} of {sku} for {event.order_id}")
@bus.handles(PaymentCaptured)
def mark_order_paid(event: PaymentCaptured) -> None:
print(f"Order {event.order_id} paid: {event.amount} ({event.processor_ref})")
class OrderService:
def place_order(self, order_id, email, items):
total = sum(price * qty for _sku, qty, price in items)
line_items = tuple((sku, qty) for sku, qty, _price in items)
# ... persist the order here ...
bus.publish(OrderPlaced(
order_id=order_id,
customer_email=email,
total=Decimal(total),
line_items=line_items,
))
return order_id
if __name__ == "__main__":
service = OrderService()
oid = service.place_order("A-1001", "buyer@example.com", [("SKU-9", 2, 15)])
bus.publish(PaymentCaptured(order_id=oid, amount=Decimal("30"), processor_ref="ch_abc"))
This snippet builds a small in-process domain event bus: events are plain immutable value objects, subscribers register handlers by event type, and a central EventBus fans out each published event to every interested handler. It is the kind of thing a service reaches for when different parts of an application need to react to "something happened" without the code that caused it knowing about the code that responds.
In events.py, each domain event is a frozen @dataclass, so instances are immutable and cheap to construct. Freezing matters: a published event is a fact about the past, and no subscriber should be able to mutate it and change what another subscriber sees. DomainEvent is a common base carrying an occurred_at timestamp, while concrete events like OrderPlaced and PaymentCaptured add their own fields. Using types as the routing key (rather than magic strings) keeps dispatch refactor-safe and lets editors autocomplete.
event_bus.py is the core. EventBus keeps a defaultdict mapping an event class to a list of handlers. subscribe registers a callable for a type, and the handles decorator is sugar over that so a function can declare its interest inline. publish looks up handlers by type(event) and calls each one; _dispatch wraps every handler in a try/except so one failing subscriber cannot stop the others, collecting failures instead. This isolation is the key trade-off of a synchronous bus — handlers run in the caller's thread, in order, but errors must be contained deliberately.
app.py shows the wiring. Handlers such as send_confirmation_email and update_inventory subscribe via @bus.handles(OrderPlaced), and the order service simply calls bus.publish(OrderPlaced(...)) after doing its work. The caller stays ignorant of who listens, which is exactly the decoupling the observer pattern buys.
The pitfalls are worth naming. This bus is synchronous and single-process, so slow handlers block the publisher and nothing survives a crash — for durability or cross-service delivery a real message queue or transactional outbox is needed. It also dispatches on exact type, not subclasses, so an OrderPlaced handler will not receive a hypothetical subclass. Within those limits it is a clean, dependency-free way to keep domain logic focused while side effects live in isolated, testable handlers.
Related snips
class Money
include Comparable
attr_reader :amount, :currency
def initialize(amount, currency = 'USD')Value objects for domain modeling
class UserRegistrationService
class Result
attr_reader :user, :errors
def initialize(success:, user: nil, errors: [])
@success = successService objects for business logic encapsulation
import { ReactNode } from 'react'
interface CardProps {
children: ReactNode
className?: string
}React component composition over inheritance
import { Controller } from "@hotwired/stimulus"
export default class extends Controller {
static targets = ["input", "results"]
static outlets = ["results-list"]
static values = {Stimulus outlets for inter-controller communication
INSTALLED_APPS += ['tenant_schemas']
DATABASE_ROUTERS = ['tenant_schemas.routers.TenantSyncRouter']
MIDDLEWARE = [
'tenant_schemas.middleware.TenantMiddleware',Django multi-tenancy with django-tenant-schemas
package cache
import (
"context"
"github.com/redis/go-redis/v9"Redis Pub/Sub subscriber with reconnect-friendly loop
Share this code
Here's the card — post it anywhere.
