Event Bus

The shift from polling to proactive delivery Most developers start by building standard REST APIs where the client asks the server for data. This polling model is fundamentally inefficient; it wastes server resources and introduces latency between an event occurring and the client discovering it. Webhooks flip this script. Instead of the client asking, the server proactively pushes data to a registered URL the moment an event happens. This architectural shift provides real-time updates and significantly reduces the load on your API infrastructure. Prerequisites and Toolkit To follow this implementation, you should have a solid grasp of Python and asynchronous programming. We utilize FastAPI for the web framework due to its high performance and Pydantic support. For sending the outgoing notifications, HTTPX serves as our asynchronous HTTP client. You will also need a basic understanding of CRUD operations and how to structure a multi-module project. Decoupling logic with an event-driven design A common mistake when adding webhooks is placing the delivery logic directly inside the API endpoint. This creates tight coupling; your core business logic suddenly needs to know about webhook payloads, URLs, and delivery status. A cleaner approach uses a lightweight Event Bus. In this pattern, the endpoint simply publishes a fact—like `link.clicked`—to the bus. The bus then notifies any subscribed listeners. This allows you to add features like logging, fraud detection, or analytics without ever touching the original endpoint code. ```python class EventBus: def __init__(self): self.listeners: dict[EventType, list[Listener]] = {} def subscribe(self, event_type: EventType, listener: Listener): self.listeners.setdefault(event_type, []).append(listener) def publish(self, event_type: EventType, data: dict): event = Event(type=event_type, data=data) for listener in self.listeners.get(event_type, []): listener(event) ``` Implementing the listener pattern Once the bus is in place, you refactor the webhook delivery into a listener. The FastAPI application initializes the bus and configures the modules. When a webhook is created, we subscribe a delivery function to the specific event type requested by the user. This ensures that the "link" module remains focused on redirects, while the "webhook" module focuses on data delivery. ```python async def deliver_webhook(webhook: Webhook, event: Event): async with httpx.AsyncClient() as client: await client.post(webhook.url, json=event.model_dump()) ``` Moving toward production stability While a local dictionary works for a tutorial, production systems require persistent storage like PostgreSQL for webhook registrations. Furthermore, sending HTTP requests during the request-response cycle can block your API. Always offload webhook delivery to a background worker or task queue like Celery or Redis. This ensures that even if a recipient's server is slow, your API remains snappy for the end user.