Skip to main content

Decoupling logic

Project description

Decouple

Say NO to monolithic architecture.

Decouple complex system to separated modules by mediator.

pip install decouple

Register event handler

register event handler

Dispatch events

dispatch events

Example

Simple usage

Code of example is here

  1. 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
  1. 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}')
  1. 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.

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Files for decouple, version 0.0.7
Filename, size File type Python version Upload date Hashes
Filename, size decouple-0.0.7.tar.gz (3.3 kB) File type Source Python version None Upload date Hashes View hashes

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging AWS AWS Cloud computing DataDog DataDog Monitoring Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page