Skip to content

EventRegistry

Keeps mappings between event types and their names.

Normally, there is no need to use it directly. If one needs to have multiple registries or wants more granular control, they can pass an instance of EventRegistry to BackendFactory.

Source code in event_sourcery/event_store/event/registry.py 
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
class EventRegistry:
    """Keeps mappings between event types and their names.

    Normally, there is no need to use it directly. If one needs to have multiple
    registries or wants more granular control, they can pass an instance
    of EventRegistry to BackendFactory."""

    def __init__(self) -> None:
        self._types_to_names: dict[type[Event], str] = {}
        self._names_to_types: dict[str, type[Event]] = {}

    def add(self, event: type["Event"]) -> type["Event"]:
        """Add event subclass to the registry."""
        if event in self._types_to_names:
            raise DuplicatedEvent(f"Duplicated Event detected! {event}")

        name = event_name(event)
        if name in self._names_to_types:
            raise DuplicatedEvent(f"Duplicated Event name detected! {name}")

        self._types_to_names[event] = name
        self._names_to_types[name] = event
        return event  # for use as a decorator

    def type_for_name(self, name: str) -> type["Event"]:
        return self._names_to_types[name]

    def name_for_type(self, event: type["Event"]) -> str:
        return self._types_to_names[event]

add(event)

Add event subclass to the registry.

Source code in event_sourcery/event_store/event/registry.py 
31
32
33
34
35
36
37
38
39
40
41
42
def add(self, event: type["Event"]) -> type["Event"]:
    """Add event subclass to the registry."""
    if event in self._types_to_names:
        raise DuplicatedEvent(f"Duplicated Event detected! {event}")

    name = event_name(event)
    if name in self._names_to_types:
        raise DuplicatedEvent(f"Duplicated Event name detected! {name}")

    self._types_to_names[event] = name
    self._names_to_types[name] = event
    return event  # for use as a decorator