How to Implement an Outbox Pattern Mock in Python

This code demonstrates a simple in-memory outbox pattern mock for publishing domain events and tracking pending events until they are marked as published.

Easy Python 3.9+ Aug 9, 2026 Microservices patterns 13 views 0 copies

outbox domain-events microservices mock dataclass

Python code

35 lines

Python 3.9+

from dataclasses import dataclass, field
from datetime import datetime
from uuid import uuid4


@dataclass
class DomainEvent:
    event_id: str = field(default_factory=lambda: str(uuid4()))
    occurred_at: datetime = field(default_factory=datetime.utcnow)


class Outbox:
    def __init__(self):
        self._events = []

    def publish(self, event: DomainEvent) -> None:
        self._events.append(event.__dict__)

    def get_pending(self) -> list[dict]:
        return list(self._events)

    def mark_published(self, event_id: str) -> None:
        self._events = [e for e in self._events if e["event_id"] != event_id]


if __name__ == "__main__":
    outbox = Outbox()
    event = DomainEvent()

    outbox.publish(event)
    print(f"Pending events: {len(outbox.get_pending())}")
    print(f"Event ID: {event.event_id}")

    outbox.mark_published(event.event_id)
    print(f"Pending after publish: {len(outbox.get_pending())}")

Output

stdout

Pending events: 1
Event ID: <uuid>
Pending after publish: 0

How it works

The DomainEvent dataclass uses a default factory to generate a unique event ID and a timestamp when the event is created. The Outbox class stores events as dictionaries in a list, simulating a database table. publish appends the event dictionary, get_pending returns a copy of the list to prevent external mutation, and mark_published filters out the event by its ID. The __main__ block publishes an event, then marks it as published, showing the count before and after. This mock is useful for testing microservice event flows without a real database.

Common mistakes

Forgetting to use `list()` when returning the events, exposing internal state to mutation.
Using `datetime.utcnow` which is deprecated in Python 3.12; prefer `datetime.now(timezone.utc)`.
Assuming `event_id` is unique; consider adding validation if needed.

Variations

Use a database table with SQLAlchemy to persist events for production.
Add a `dispatch_domain_events` function to send events to a message broker like RabbitMQ or Kafka.

Real-world use cases

Unit-testing event-driven microservices to verify domain events are published and eventually dispatched.
Building a local development environment that simulates transactional outbox behavior without external infrastructure.
Prototyping a transactional outbox implementation before choosing a message broker and database.

How to Implement an Outbox Pattern Mock in Python

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Microservices patterns

Tutorials

Quizzes

Python code

Output

How it works

Common mistakes

Variations

Real-world use cases

More from Microservices patterns

Keep learning

Tutorials

Quizzes