Event-Driven Python with pypubsub

5 min readSep 25, 2022

Simple Event Driven Python Example — Explained

We will use a fictious ordering system to demonstrate the implementation of event-driven system in Python.

We need to implement a workflow with following steps: 1. Order is placed by a customer 2. Order is created by persisting it to the database 3. Sales department receives a notfication about the new order

All the source for this example is available in GitHub.

Let’s start by installing the pypubsub package:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">$ pip install pypubsub

Our order is very simple — just a list items. Each item is represented by a string:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">items = ["100 inch TV", "magic carpet"]

To store the order we need a bit more complex strucure. We use a dictionary with two items: order id and order items. We are using uuid4() to generate unique order id.

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">order = {
    "id": uuid4(),
    "items": ["100 inch TV", "magic carpet"]
}

As database we use Python dictionary. As dictionary key we use the order id:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">orders: Dict[UUID, dict] = {}

Saving an order to the database is as simple as adding a key to the orders dictionary.

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">orders[order["id"]] = order

Our system uses pubsub to orchestrate messages with two topics:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">PLACE_ORDER_TOPIC = 'place-order'
ORDER_CREATED_TOPIC = 'order-created'

When an order is placed by a customer, we want to call a service function which takes care about creating a new order object and storing it into the database. Once the order is created, an event is triggered to notify other parts of the system:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">def place_order(items: List[str]):
    order = Order(items.copy())
    orders[order.id] = order
    pub.sendMessage(ORDER_CREATED_TOPIC, order_id=order.id)

We need to subscribe the place_order handler to the PLACE_ORDER_TOPIC command topic:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">pub.subscribe(place_order, PLACE_ORDER_TOPIC)

To notify the sales department about new orders, we are listening for events on the ORDER_CREATED_TOPIC:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">def notify_sales(message: str):
    print(message)def order_created(order_id: UUID):
    order = orders[order_id]
    notify_sales(f"============\nNew Order:\n============\nID: {order['id']}\nItems: {order['items']}")pub.subscribe(order_created, ORDER_CREATED_TOPIC)

Publishing a message to the PLACE_ORDER_TOPIC will kick our process: - Trigger the place_order handler which will store the order and place a message to the ORDER_CREATED_TOPIC - Trigger the order_created handler which will call the notify_sales service function

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">pub.sendMessage(PLACE_ORDER_TOPIC, items=["100 inch TV", "magic carpet"])

This will produce following output:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">============
New Order:
============
ID: aafc3ad9-417c-484b-8ed7-3960e7241ba8
Items: ['100 inch TV', 'magic carpet']

Simple Event Driven Python Example — Full Code

Below is the full source code (file simple_pubsub.py) for our simple event-driven sales system in Python:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--lightest); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">from dataclasses import dataclass, field
from typing import Dict, List
from uuid import UUID, uuid4
from pubsub import pubPLACE_ORDER_TOPIC = 'place-order'
ORDER_CREATED_TOPIC = 'order-created'orders: Dict[UUID, 'Order'] = {}
def place_order(items: List[str]):
    order = Order(items.copy())
    orders[order.id] = order
    pub.sendMessage(ORDER_CREATED_TOPIC, order_id=order.id)def notify_sales(message: str):
    print(message)def order_created(order_id: UUID):
    order = orders[order_id]
    notify_sales(f"============\nNew Order:\n============\nID: {order.id}\nItems: {order.items}")pub.subscribe(place_order, PLACE_ORDER_TOPIC)
pub.subscribe(order_created, ORDER_CREATED_TOPIC)pub.sendMessage(PLACE_ORDER_TOPIC, items=["100 inch TV", "magic carpet"])
pub.sendMessage(PLACE_ORDER_TOPIC, items=["soft cheese", "dutch mashrooms"])

Running the simple_pubsub.py produces two notifications about placed orders:

code" style="box-sizing: inherit; -webkit-tap-highlight-color: transparent; background: transparent; border: 0px; font-family: inherit; font-size: inherit; margin: 0px; padding: 0px; border-radius: 0.1rem; color: var(--md-default-fg-color--light); cursor: pointer; height: 1.5em; outline: none; outline-offset: 0.1rem; position: absolute; right: 0.5em; top: 0.5em; transition: color 0.25s ease 0s; width: 1.5em; z-index: 1;">============
New Order:
============
ID: aafc3ad9-417c-484b-8ed7-3960e7241ba8
Items: ['100 inch TV', 'magic carpet']
============
New Order:
============
ID: 6d9be0ec-6671-43d8-81c7-a2aa317d375c
Items: ['soft cheese', 'dutch mashrooms']

This was originally posted on GitHub pages where you could also find advanced implementation of the same scenario: Event-Driven Python with pubsub.

Event-Driven Python with pypubsub

Simple Event Driven Python Example — Explained

Simple Event Driven Python Example — Full Code

Written by Ivan Georgiev