event-driven

event-driven

Covers how to design, implement, and operate event-driven systems: pub/sub messaging, event buses, async pipelines, and reactive coordination between services.

1) Redis pub/sub fundamentals

import redis.asyncio as aioredis
import asyncio, json

redis = aioredis.from_url("redis://localhost:6379", decode_responses=True)

# Publisher
async def publish(channel: str, event: dict):
    await redis.publish(channel, json.dumps({**event, "_ts": asyncio.get_event_loop().time()}))

# Subscriber (persistent listener)
async def subscribe(channels: list[str], handler):
    async with redis.pubsub() as pubsub:
        await pubsub.subscribe(*channels)
        async for message in pubsub.listen():
            if message["type"] == "message":
                await handler(json.loads(message["data"]))

2) Event bus abstraction

from collections import defaultdict
from typing import Callable, Awaitable

EventHandler = Callable[[dict], Awaitable[None]]

class EventBus:
    def __init__(self):
        self._handlers: dict[str, list[EventHandler]] = defaultdict(list)

    def on(self, event_type: str):
        """Decorator to register a handler."""
        def decorator(fn: EventHandler) -> EventHandler:
            self._handlers[event_type].append(fn)
            return fn
        return decorator

    async def emit(self, event_type: str, payload: dict):
        for handler in self._handlers.get(event_type, []):
            await handler({"type": event_type, **payload})

    async def emit_all(self, events: list[tuple[str, dict]]):
        """Emit multiple events concurrently."""
        await asyncio.gather(*[self.emit(t, p) for t, p in events])

bus = EventBus()

@bus.on("article.published")
async def on_article_published(event: dict):
    await index_article(event["slug"])

@bus.on("article.published")
async def notify_subscribers(event: dict):
    await send_notification(event["slug"])

3) Durable event queue (Redis Streams)