A Simple Factory for Domain Events

This is a simple demonstration of a domain event factory in Python. I assume you are familiar with the Factory Method Pattern. I also use the pydantic package for attribute validation. When implemented, we can use the factory to create immutable domain events with a homogenous data structure across instances of the same type. The metadata is generated by the underlying BaseEvent. In this approach we always produces complete events.

Start with all our imports:

from datetime import datetime
from typing import Union, List
from uuid import UUID, uuid4

from pydantic import BaseModel
from pydantic.fields import Field, ModelField

Base Event class

the BaseEvent includes metadata ( id and created_at ) for all events. The BaseModel.Config sets the object to be immutable and no extra arguments are allowed:

class EventMetaData(BaseModel):
    created_at: datetime = Field(default_factory=datetime.now)
    class Config:
        allow_mutation = False
        extra = "forbid"

class BaseEvent(BaseModel):
    """Generic domain event."""

    class Config:
        allow_mutation = False
        extra = "forbid"

    id: UUID = Field(default_factory=uuid4)
    metadata: EventMetaData = Field(default_factory=EventMetaData)

we introduce a method that returns a new event type from the BaseEvent with a name and an additional datastructure:

def create_event_type(
    event_type: str,
    data_structure: dict,
    doc_string: Union[str, None] = None,
):
    """Return new domain event type."""
    _cls = type(event_type, (BaseEvent,), {})
    _cls.__annotations__.update(data_structure)
    if doc_string:
        _cls.__doc__ = doc_string
    _fields: Dict[str, ModelField] = {}
    for fname, f_def in data_structure.items():
        f_val = ...
        if isinstance(f_def, tuple):
            f_def, f_val = f_def
        _fields[fname] = ModelField.infer(
            name=fname,
            value=f_val,
            annotation=f_def,
            class_validators=None,
            config=_cls.__config__,
        )
    _cls.__fields__.update(_fields)
    return _cls

A simle demonstration of the create_event function:

>>> CommentWritten = create_event_type("CommentWritten", {"author": str})
>>> CommentWritten(author="joost").json()
'{"id": "85990e7e-9705-4450-849e-70a764593e81", "metadata": {"created_at": "2021-01-25T11:45:19.998831"}, "author": "joost"}'

The EventTypeFactory

The factory uses the create_event_type function to register Domain Event Type classes:

class EventTypeFactory:
    def __init__(self):
        self._event_types = {}
        self._id_generator = lambda: None

    def register_event_type(self, event_type: str, data_structure: dict):
        self._event_types[event_type] = create_event_type(
            event_type, data_structure, self._id_generator
        )

    def get_event_type(self, event_type):
        event_type = self._event_types.get(event_type)
        if not event_type:
            raise ValueError(event_type)
        return event_type

Now we can register EventTypes with a data structure:

>>> factory = EventTypeFactory()
>>> factory.register_event_type(
    "BoughtWineEvent",
    {"year": int, "wine": str},
)
>>> factory.register_event_type(
    "AnotherEvent",
    {"ready": bool},
)

and use the factory to get event type classes:

>>> BoughtWineEvent = factory.get_event_type("BoughtWineEvent")
>>> print("type of event:", type(BoughtWineEvent))
type of event: <class 'pydantic.main.ModelMetaclass'>

>>> event = BoughtWineEvent(
    year=2008,
    wine="Pinot Noir",
)
>>> event.json()
'{"id": "f880be5c-d352-4149-abfa-e5f02a61eb92", "metadata": {"created_at": "2021-01-25T11:46:09.222415"}, "year": 2008, "wine": "Pinot Noir"}'

I’m still considering nesting the metadata and data in designated keys but otherwise the factory works perfectly :)

Base Event class#

The EventTypeFactory#

Base Event class

The EventTypeFactory