python 34 lines · 8 steps

A pattern-matching event bus in Python

A publish/subscribe hub that routes events to handlers via glob patterns and doubles as a decorator.

Explained by highlit
Intermediate pub-sub decorators pattern-matching defaultdict callbacks
1import fnmatch
2from collections import defaultdict
3from typing import Any, Callable
4 
5 
6class EventBus:
7 def __init__(self) -> None:
8 self._handlers: dict[str, list[Callable[..., Any]]] = defaultdict(list)
9 
10 def subscribe(self, pattern: str, handler: Callable[..., Any] | None = None):
11 if handler is None:
12 def decorator(fn: Callable[..., Any]) -> Callable[..., Any]:
13 self._handlers[pattern].append(fn)
14 return fn
15 return decorator
16 self._handlers[pattern].append(handler)
17 return handler
18 
19 def unsubscribe(self, pattern: str, handler: Callable[..., Any]) -> None:
20 listeners = self._handlers.get(pattern)
21 if listeners and handler in listeners:
22 listeners.remove(handler)
23 if not listeners:
24 del self._handlers[pattern]
25 
26 def publish(self, event: str, *args: Any, **kwargs: Any) -> int:
27 delivered = 0
28 for pattern, listeners in list(self._handlers.items()):
29 if not fnmatch.fnmatchcase(event, pattern):
30 continue
31 for handler in list(listeners):
32 handler(event, *args, **kwargs)
33 delivered += 1
34 return delivered
01 / 01
STEP 01

Walkthrough

Space play ←→ step click any line
Three takeaways
  1. 1A defaultdict of lists lets you register handlers under a key without existence checks.
  2. 2Making the handler argument optional turns one method into both a direct call and a decorator.
  3. 3Iterating over copies of mutable collections keeps publish safe against handlers that subscribe or unsubscribe mid-dispatch.

Related explainers

python 
from fastapi import FastAPI, WebSocket, WebSocketDisconnect
 
app = FastAPI()
 

Building a WebSocket chat with FastAPI

websockets broadcast connection-management
Intermediate 9 steps
ruby 
class UserAgentParser
  BROWSERS = [
    [/Edg\/([\d.]+)/, "Edge"],
    [/OPR\/([\d.]+)/, "Opera"],

Parsing user-agent strings in Ruby

regex pattern-matching lookup-tables
Intermediate 8 steps
python 
import time
import uuid
 
from django.utils.deprecation import MiddlewareMixin

Attaching per-request context in Django

middleware request lifecycle multi-tenancy
Intermediate 7 steps
python 
import random
from typing import Iterator, List
 
 

How reservoir sampling picks k items

reservoir-sampling streaming randomness
Intermediate 5 steps
python 
import secrets
 
from django.contrib.auth import authenticate, login
from django.core.cache import cache

Two-factor login with OTP in Django

two-factor-auth one-time-passwords caching
Intermediate 9 steps
python 
import re
from functools import total_ordering
from typing import Optional
 

Parsing and comparing semantic versions

regex operator-overloading sorting
Intermediate 7 steps

Share this explainer

Here's the card — post it anywhere.

A pattern-matching event bus in Python — share card
Made with highlit — turn any snippet into a walkthrough like this in about a minute.
Explain your code