Introduction
I enjoy refactoring code after a period of deep learning – it is often when my biggest architectural realisations land. Coming back to MetaMaker, a Python desktop application originally built around file metadata and Darkblock1 upload workflows for an experimental steganographic art collection of my own paintings, I saw room to refine how its components communicated internally.
MetaMaker used an event-driven MVC architecture. To decouple components and standardise event handling, I introduced Blinker2, a lightweight Python library for dispatching signals. This post focuses on how I hardened that signal layer: introducing explicit payload classes, enforcing runtime payload contracts, and adopting a hybrid object + ID pattern to balance responsiveness against reliability.
The main MetaMaker case study covers the wider architecture; this post focuses specifically on how the event layer evolved. Coming from a Java background, the process also helped me settle into a more Pythonic and maintainable approach to event-driven design.
Introducing Event Payloads
Early in the Blinker rollout, signal emissions were simple:
1
set_manager_on_updated.send(self, payload=nft_set)
The trouble was structure. Objects were passed informally, with no defined schema and no type safety. As the system grew, that ambiguity made payloads harder to validate, handlers harder to test, and signal flow harder to trace across components.
To address this, I introduced explicit payload classes using @dataclass3:
1
2
3
4
5
6
7
8
@dataclass
class SetUpdatedEvent:
nft_set: 'NftMetadataSet'
set_id: int
def emit_set_updated(sender, nft_set):
payload = SetUpdatedEvent(nft_set=nft_set, set_id=nft_set.id)
set_manager_on_updated.send(sender, payload=payload)
Benefit: handlers can access fields predictably (payload.set_id, payload.nft_set.name), and the code becomes easier to read and refactor.
Formalising events this way led me to a broader architectural question: should I really be passing full objects around, and was I violating the “single source of truth” principle by doing so?
The “Single Source of Truth” Principle
Conventionally, data should be re-fetched from the database (the repository) to ensure it is current, especially in multi-threaded4 or distributed systems.
In MetaMaker’s case, that convention did not always make sense. When emitting a signal immediately after writing to the database, I already had the “truth” in memory. Forcing handlers to re-query that data introduced latency and redundant work – defeating the responsiveness goals of the UI layer.
The hybrid approach struck a balance:
- IDs preserve the ability to re-fetch the authoritative state when needed.
- Object references allow fast, contextual responses in the UI – without assuming they are immutable or canonical.
- Event classes (like
SetUpdatedEvent) formalise this structure, making the trade-off clear and auditable.
Takeaway:
Convention is a guide and not a rule. When you deviate from it, the key is to do so transparently – with structure and safeguards in place.
The Hybrid Payload Model: Object + ID
By wrapping both the object and its ID in a structured dataclass (e.g. DblockUpdatedEvent), I gained:
- Immediate UI access to object data (
payload.nft_set.name) - Optional DB refresh when needed (
darkblockmanager.get_dblock(dblockid=42)5) - Runtime payload contracts for debugging and handler validation
- Cleaner testing, using mock event payloads and signal assertions
This resolved several pain points at once. It preserved the responsiveness of object-based signalling while introducing the structure and reliability needed to scale. With hybrid payloads, handlers can now:
- Update UI elements immediately from object attributes (
payload.nft_set.name) - Fetch fresh state only when needed (
darkblockmanager.get_dblock(dblockid=42))
Takeaway:
The hybrid model offers flexibility: immediate access when speed matters and strict consistency when correctness does.
Enforcing Runtime Payload Contracts
To guard against accidental signal misuse, I added a decorator that validates payload types when an environment flag is set:
1
2
3
4
5
6
7
8
9
10
11
def enforce_contract(signal_name, expected_type):
def decorator(func):
@wraps(func)
def wrapper(sender, **kwargs):
if os.getenv("DEBUG_SIGNALS") == "1":
payload = kwargs.get('payload')
if not isinstance(payload, expected_type):
raise TypeError(f"Signal '{signal_name}' expected {expected_type.__name__}")
return func(sender, **kwargs)
return wrapper
return decorator
This let me catch mistakes safely during development, with no production overhead. To reinforce it, I also temporarily patched Signal.send to flag any legacy manual emissions:
1
2
3
4
5
6
7
8
9
10
11
12
13
def patched_send(self, *args, **kwargs):
stack = inspect.stack()
if not any("emit_" in frame.function for frame in stack):
caller = next((f for f in stack if "emit_" not in f.function), None)
if caller:
logger.warning(
f"[LEGACY SIGNAL USE] Signal '{self.name}' sent manually from: \n"
f"{caller.filename}:{caller.lineno}; \n Refactor using emitter and dataclass payloads."
)
return _original_send(self, *args, **kwargs)
if os.getenv("DEBUG_SIGNALS", "1") == "1":
Signal.send = patched_send
Takeaway:
This log snapshot captures the lifecycle of a view initialisation. It surfaces a legacy signal warning – thanks to monkey-patching Signal.send – and shows a TypeError enforcing the expected payload structure.
Testing Strategy
To keep the system reliable, I settled on a consistent testing approach for signals.
Use fake payloads
1
2
3
4
class FakeNftMetadataSet:
def __init__(self, id):
self.id = id
self.name = "Test"
Patch the environment
1
2
3
4
5
6
invalid_payloads = [("string", "invalid"), ("none", None)]
with patch.dict('os.environ', {'DEBUG_SIGNALS': '1'}):
for case_name, payload in invalid_payloads:
with self.subTest(case=case_name):
with self.assertRaises(TypeError):
handler(None, payload=payload)
This enforces signal contracts during testing, with minimal noise and clear subtest isolation.
Clean up handlers
Handlers persist across tests unless removed:
1
2
3
4
5
def tearDown(self):
for signal in self.ALL_SIGNALS:
for receiver in signal.receivers.values():
signal.disconnect(receiver)
self.handlers.clear()
Together, these steps made the signal system reliable and testable, even during concurrency-heavy operations.
Looking Back
MetaMaker is now archived as a portfolio project, so I would not continue this migration for its own sake. If I were rebuilding the architecture today, I would either standardise the remaining listener-based paths around Blinker or deliberately keep them separate as a smaller async-result boundary.
The important lesson was not “use Blinker everywhere.” It was that event-driven code needs structure. Named payloads, explicit emit functions, runtime contracts, and predictable cleanup turned a loose signal layer into something easier to test, trace, and refactor.
MetaMaker – Next Steps
For the full picture on this project, read the MetaMaker case study, or browse the code on GitHub.
Or, since you have read this far – a choice:
Let’s Connect
If you are interested in event-driven Python architecture, or want to talk about signal layers, typed payloads, or testing event-driven systems, I’d be glad to hear from you.
Prefer to email me directly? hello@lundie.io
Thanks – your message is on its way. I'll get back to you soon.
Something went wrong sending that. Please try again, or email me directly at hello@lundie.io.
-
Darkblock is a service for attaching encrypted, unlockable content to digital assets. ↩
-
Event-Driven Programming is a design pattern where components react to events; Blinker is a Python library that enables this through signals. ↩
-
@dataclassis a decorator in Python that automatically generates special methods like__init__,__repr__, and__eq__for classes used primarily to store data. ↩ -
Multithreading allows multiple tasks to run concurrently, improving responsiveness in applications like MetaMaker. ↩
-
DarkblockManager is a custom manager class responsible for updating the database with Darkblock transaction data and emitting signals. ↩