Skip to content

Migration event manager

mongorunway.domain.migration_event_manager ¤

__all__: typing.Sequence[str] = ('MigrationEventManager') module-attribute ¤

MigrationEventManager ¤

Bases: abc.ABC

Source code in mongorunway\domain\migration_event_manager.py 
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 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
class MigrationEventManager(abc.ABC):
    __slots__: typing.Sequence[str] = ()

    @abc.abstractmethod
    def subscribe_events(
        self,
        handler: domain_event.EventHandlerProxyOr[domain_event.EventHandler],
        *events: typing.Type[domain_event.MigrationEvent],
    ) -> None:
        ...

    @abc.abstractmethod
    def unsubscribe_events(self, event: typing.Type[domain_event.MigrationEvent]) -> None:
        ...

    @abc.abstractmethod
    def subscribe_event_handler(
        self,
        handler: domain_event.EventHandlerProxyOr[domain_event.EventHandler],
        event: typing.Type[domain_event.MigrationEvent],
    ) -> None:
        ...

    @abc.abstractmethod
    def unsubscribe_event_handler(
        self,
        handler: domain_event.EventHandlerProxyOr[domain_event.EventHandler],
        event: typing.Type[domain_event.MigrationEvent],
    ) -> None:
        ...

    @abc.abstractmethod
    def get_event_handlers_for(
        self,
        event: typing.Type[domain_event.MigrationEvent],
    ) -> typing.MutableSequence[domain_event.EventHandlerProxyOr[domain_event.EventHandler]]:
        ...

    @abc.abstractmethod
    def prioritize_handler(
        self,
        handler: domain_event.EventHandler,
        event: typing.Type[domain_event.MigrationEvent],
        priority: int,
    ) -> None:
        ...

    @abc.abstractmethod
    def unprioritize_handler_proxy(
        self,
        handler_proxy: domain_event.EventHandlerProxy,
        event: typing.Type[domain_event.MigrationEvent],
    ) -> None:
        ...

    @abc.abstractmethod
    def listen(
        self,
        *events: typing.Type[domain_event.MigrationEvent],
    ) -> typing.Callable[
        [domain_event.EventHandlerProxyOr[domain_event.EventHandlerT]],
        domain_event.EventHandlerProxyOr[domain_event.EventHandlerT],
    ]:
        ...

    @abc.abstractmethod
    def dispatch(self, event: domain_event.MigrationEvent) -> None:
        ...

__slots__: typing.Sequence[str] = () instance-attribute class-attribute ¤

dispatch(event) abstractmethod ¤

Source code in mongorunway\domain\migration_event_manager.py 
 99
100
101
@abc.abstractmethod
def dispatch(self, event: domain_event.MigrationEvent) -> None:
    ...

get_event_handlers_for(event) abstractmethod ¤

Source code in mongorunway\domain\migration_event_manager.py 
65
66
67
68
69
70
@abc.abstractmethod
def get_event_handlers_for(
    self,
    event: typing.Type[domain_event.MigrationEvent],
) -> typing.MutableSequence[domain_event.EventHandlerProxyOr[domain_event.EventHandler]]:
    ...

listen(*events) abstractmethod ¤

Source code in mongorunway\domain\migration_event_manager.py 
89
90
91
92
93
94
95
96
97
@abc.abstractmethod
def listen(
    self,
    *events: typing.Type[domain_event.MigrationEvent],
) -> typing.Callable[
    [domain_event.EventHandlerProxyOr[domain_event.EventHandlerT]],
    domain_event.EventHandlerProxyOr[domain_event.EventHandlerT],
]:
    ...

prioritize_handler(handler, event, priority) abstractmethod ¤

Source code in mongorunway\domain\migration_event_manager.py 
72
73
74
75
76
77
78
79
@abc.abstractmethod
def prioritize_handler(
    self,
    handler: domain_event.EventHandler,
    event: typing.Type[domain_event.MigrationEvent],
    priority: int,
) -> None:
    ...

subscribe_event_handler(handler, event) abstractmethod ¤

Source code in mongorunway\domain\migration_event_manager.py 
49
50
51
52
53
54
55
@abc.abstractmethod
def subscribe_event_handler(
    self,
    handler: domain_event.EventHandlerProxyOr[domain_event.EventHandler],
    event: typing.Type[domain_event.MigrationEvent],
) -> None:
    ...

subscribe_events(handler, *events) abstractmethod ¤

Source code in mongorunway\domain\migration_event_manager.py 
37
38
39
40
41
42
43
@abc.abstractmethod
def subscribe_events(
    self,
    handler: domain_event.EventHandlerProxyOr[domain_event.EventHandler],
    *events: typing.Type[domain_event.MigrationEvent],
) -> None:
    ...

unprioritize_handler_proxy(handler_proxy, event) abstractmethod ¤

Source code in mongorunway\domain\migration_event_manager.py 
81
82
83
84
85
86
87
@abc.abstractmethod
def unprioritize_handler_proxy(
    self,
    handler_proxy: domain_event.EventHandlerProxy,
    event: typing.Type[domain_event.MigrationEvent],
) -> None:
    ...

unsubscribe_event_handler(handler, event) abstractmethod ¤

Source code in mongorunway\domain\migration_event_manager.py 
57
58
59
60
61
62
63
@abc.abstractmethod
def unsubscribe_event_handler(
    self,
    handler: domain_event.EventHandlerProxyOr[domain_event.EventHandler],
    event: typing.Type[domain_event.MigrationEvent],
) -> None:
    ...

unsubscribe_events(event) abstractmethod ¤

Source code in mongorunway\domain\migration_event_manager.py 
45
46
47
@abc.abstractmethod
def unsubscribe_events(self, event: typing.Type[domain_event.MigrationEvent]) -> None:
    ...