Decouple
Say NO to monolithic architecture.
Decouple complex system to separated modules by mediator.
pip install decoupleRegister event handler
Dispatch events
Example
Simple usage
Code of example is here
- Write ModuleA - publisher
from dataclasses import dataclass
from decouple import Module, Event, Mediator
class ModuleA(Module):
def __init__(self, mediator: Mediator = Mediator()):
super().__init__(mediator)
def start(self):
self.pub(StartEvent(a=7))
@dataclass
class StartEvent(Event):
a: int = 0 # must be serializable- Write ModuleB - subscriber
from decouple import Module
from examples.simple.module_a import StartEvent
class ModuleB(Module):
def __init__(self):
super().__init__()
self.sub(StartEvent, self.handle_a)
def handle_start(self, event: StartEvent):
print(f'field a:{event.a}')
- Compose both modules to the whole
from examples.simple.module_a import ModuleA
from examples.simple.module_b import ModuleB
module_a = ModuleA()
module_a.add(ModuleB())
module_a.start()Priorities
Manual control
from decouple import Module
from examples.simple.module_a import StartEvent
class ModuleB(Module):
def __init__(self):
super().__init__()
# handler with higher priority will be triggered early
self.sub(StartEvent, self.handle_b, priority=0)
self.sub(StartEvent, self.handle_a, priority=100)
def handle_a(self, event: StartEvent):
# will be triggered first
print(f'field a:{event.a}')
def handle_b(self, event: StartEvent):
# will be triggered second
print(f'event.uuid:{event.uuid}, event.timestamp:{event.timestamp}')Default priority depends on registration order
from decouple import Module
from examples.simple.module_a import StartEvent
class ModuleB(Module):
def __init__(self):
super().__init__()
# priority of handlers call increased by 1 every registration for the same event
self.sub(StartEvent, self.handle_b) # priority=1
self.sub(StartEvent, self.handle_a) # priority=2
def handle_a(self, event: StartEvent):
# will be triggered second
print(f'field a:{event.a}')
def handle_b(self, event: StartEvent):
# will be triggered first
print(f'event.uuid:{event.uuid}, event.timestamp:{event.timestamp}')Feedback
I will be glad to read your feedback in issues and pull requests.
