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 
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
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[TEvent], str] = {}
        self._names_to_types: dict[str, type[TEvent]] = {}
        self._encrypted_fields: dict[type[TEvent], dict[str, Encrypted]] = {}
        self._subject_fields: dict[type[TEvent], str] = {}
        self._register_defined_events()

    def _register_defined_events(self) -> None:
        for event_type in Event.__subclasses__():
            not_registered_type = event_type not in self._types_to_names
            not_registered_name = event_name(event_type) not in self._names_to_types
            not_registered_type and not_registered_name and self.add(event_type)

    def add(self, event: type[TEvent]) -> type[TEvent]:
        """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
        self._encrypted_fields[event] = get_encrypted_fields(of=event)
        subject_field = get_data_subject_filed(of=event)
        if subject_field is not None:
            self._subject_fields[event] = subject_field
        return event  # for use as a decorator

    def type_for_name(self, name: str) -> type[TEvent]:
        if name not in self._names_to_types:
            self._register_defined_events()
        return self._names_to_types[name]

    def name_for_type(self, event: type[TEvent]) -> str:
        if event not in self._types_to_names:
            self._register_defined_events()
        return self._types_to_names[event]

    def encrypted_fields(self, of: type[TEvent]) -> dict[str, Encrypted]:
        return self._encrypted_fields[of]

    def subject_filed(self, for_field: str, of: type[TEvent]) -> str | None:
        metadata = self._encrypted_fields[of].get(for_field)
        return (metadata and metadata.subject_field) or self._subject_fields.get(of)

add(event)

Add event subclass to the registry.

Source code in event_sourcery/event_store/event/_registry.py 
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
def add(self, event: type[TEvent]) -> type[TEvent]:
    """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
    self._encrypted_fields[event] = get_encrypted_fields(of=event)
    subject_field = get_data_subject_filed(of=event)
    if subject_field is not None:
        self._subject_fields[event] = subject_field
    return event  # for use as a decorator