Quick Start¶

In this quick start, we will build a simple message driven system using Mersal. It should only take a few minutes. We won’t use a production ready setup nor will we dive deep into how things work. This is just an appetizer. If you are already familiar with message driven systems or want a deeper dive, then one of the tutorials would be a better fit.

No more e-commerce examples please

If you open a random blog post or tutorial or some sample source code that explains a topic related to distributed systems or message/event driven architecture, the most likely scenario you will find is an e-commerce example. If you went through a lot of these, you will probably say “oh not e-commerce again” once you start reading our quick start guide here.

We agree, it has become a boring example (despite being very challenging and an excellent use case for demonstration). To avoid this cliche, we promise that beyond the quick start guide, we will not use an e-commerce scenario (see Example Theme ). So, e-commerce scenario one more time only!

The scenario¶

An online store lets shoppers save products to a wishlist. When a product’s price drops, everyone who has that product on their wishlist should get an email.

We promised to make this a quick start guide so we won’t discuss solution approaches whether it’s message driven or not so lets jump right in.

Installation¶

Lets create the virtual environment, assuming uv is used:

uv init mersal-quickstart

Now lets add Mersal as a dependency

cd mersal-quickstart

uv add mersal

The domain¶

We define the two messages that will be used in our system. They can be defined using two plain dataclasses.

Lets create a file called messages.py and place the following into it:

messages.py¶

from dataclasses import dataclass

__all__ = (
    "PriceDecreased",
    "UpdateProductPrice",
)


@dataclass
class UpdateProductPrice:
    product_id: str
    new_price: float


@dataclass
class PriceDecreased:
    product_id: str
    new_price: float

Mersal Apps¶

Lets think of our system as it’s made up of multiple app. Each app is responsible for handling a certain message. A message can also be handled by multiple apps. We will call each of these as a “Mersal app”.

For this system, we will have two Mersla apps, one of them will be called “catalog” and the other one will be called “notifications”. Lets create them now.

Create a new file called main.py

main.py¶

from __future__ import annotations

import argparse
from pathlib import Path

import anyio
from notify_wishlist_users_handler import NotifyWishlistUsersHandler
from record_price_change_handler import RecordPriceChangeHandler
from update_price_message_handler import UpdateProductPriceHandler

from mersal.activation import BuiltinHandlerActivator
from mersal.app import Mersal
from mersal.persistence.in_memory import (
    InMemorySubscriptionStorage,
    InMemorySubscriptionStore,
)
from mersal.serialization.dataclass_serializer import DataclassSerializer
from mersal.transport.file_system import FileSystemTransportConfig
from messages import PriceDecreased, UpdateProductPrice

__all__ = (
    "main",
    "make_mersal_app",
)


def make_mersal_app(
    name: str,
    base_directory: str | Path,
    subscription_store: InMemorySubscriptionStore | None = None,
) -> tuple[Mersal, BuiltinHandlerActivator]:
    activator = BuiltinHandlerActivator()
    transport = FileSystemTransportConfig(
        base_directory=base_directory,
        input_queue_address=name,
    ).transport
    if subscription_store is None:
        subscription_store = InMemorySubscriptionStore()
    app = Mersal(
        name,
        activator,
        transport=transport,
        message_body_serializer=DataclassSerializer({UpdateProductPrice, PriceDecreased}),
        subscription_storage=InMemorySubscriptionStorage.centralized(subscription_store),
    )
    return app, activator


async def main(run_catalog: bool, run_notifications: bool) -> None:
    subscription_store = InMemorySubscriptionStore()
    base_directory = ".mersal-quickstart"

    catalog_app, catalog_activator = make_mersal_app("catalog", base_directory, subscription_store)
    notifications_app, notifications_activator = make_mersal_app("notifications", base_directory, subscription_store)
    accounting_app, accounting_activator = make_mersal_app("accounting", base_directory, subscription_store)

    catalog_activator.register(UpdateProductPrice, lambda _, app: UpdateProductPriceHandler(app))
    notifications_activator.register(PriceDecreased, lambda _, __: NotifyWishlistUsersHandler())
    accounting_activator.register(PriceDecreased, lambda _, __: RecordPriceChangeHandler())
    await notifications_app.subscribe(PriceDecreased)
    await accounting_app.subscribe(PriceDecreased)

    if run_catalog:
        await catalog_app.start()
    if run_notifications:
        await notifications_app.start()
    await accounting_app.start()

    print("Running. Press Ctrl+C to stop.")
    await anyio.sleep_forever()


if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--catalog", action="store_true")
    parser.add_argument("--notifications", action="store_true")
    args = parser.parse_args()

    run_both = not args.catalog and not args.notifications
    anyio.run(main, args.catalog or run_both, args.notifications or run_both)

Handling the update price message¶

Right now the two apps exist but neither one knows how to handle any message. Lets fix the catalog side first:

Create a new file called update_price_message_handler.py

from __future__ import annotations

from typing import TYPE_CHECKING

from messages import PriceDecreased, UpdateProductPrice

if TYPE_CHECKING:
    from mersal.app import Mersal

__all__ = ("UpdateProductPriceHandler",)


_catalog_prices: dict[str, float] = {}


class UpdateProductPriceHandler:
    def __init__(self, catalog_app: Mersal) -> None:
        self._app = catalog_app

    async def __call__(self, command: UpdateProductPrice) -> None:
        _catalog_prices[command.product_id] = command.new_price
        print(f"[catalog] {command.product_id} is now ${command.new_price:.2f}")
        await self._app.publish(PriceDecreased(command.product_id, command.new_price))

The catalog’s handler updates the catalog prices (in production this would be a database call), then publishes the event.

Now we need to update main.py so that the catalog Mersal app can handle that message. Register the handler on the catalog’s activator, inside main():

main.py¶

async def main(run_catalog: bool, run_notifications: bool) -> None:
    subscription_store = InMemorySubscriptionStore()
    base_directory = ".mersal-quickstart"

    catalog_app, catalog_activator = make_mersal_app("catalog", base_directory, subscription_store)
    notifications_app, notifications_activator = make_mersal_app("notifications", base_directory, subscription_store)
    accounting_app, accounting_activator = make_mersal_app("accounting", base_directory, subscription_store)

    catalog_activator.register(UpdateProductPrice, lambda _, app: UpdateProductPriceHandler(app))
    notifications_activator.register(PriceDecreased, lambda _, __: NotifyWishlistUsersHandler())
    accounting_activator.register(PriceDecreased, lambda _, __: RecordPriceChangeHandler())
    await notifications_app.subscribe(PriceDecreased)
    await accounting_app.subscribe(PriceDecreased)

    if run_catalog:
        await catalog_app.start()
    if run_notifications:
        await notifications_app.start()
    await accounting_app.start()

    print("Running. Press Ctrl+C to stop.")
    await anyio.sleep_forever()

Reacting to the event¶

The notifications side only needs to know about PriceDecreased. It has no idea UpdateProductPrice exists, and the catalog has no idea this handler exists either - that’s the whole point of pub/sub.

Create a new file called notify_wishlist_users_handler.py:

notify_wishlist_users_handler.py¶

from __future__ import annotations

from messages import PriceDecreased

__all__ = (
    "FlakyNotifyWishlistUsersHandler",
    "NotifyWishlistUsersHandler",
)


_wishlists: dict[str, list[str]] = {
    "product-42": ["[email protected]", "[email protected]"],
}


class NotifyWishlistUsersHandler:
    async def __call__(self, event: PriceDecreased) -> None:
        for email in _wishlists.get(event.product_id, []):
            print(f"[notifications] Emailing {email}: {event.product_id} dropped to ${event.new_price:.2f}")


class FlakyNotifyWishlistUsersHandler:
    def __init__(self) -> None:
        self.attempts = 0

    async def __call__(self, event: PriceDecreased) -> None:
        self.attempts += 1
        print(f"[notifications] Attempt {self.attempts}: emailing wishlist for {event.product_id}...")
        raise ConnectionError("SendGrid API is currently unreachable!")

Register it on the notifications app, and subscribe that app to PriceDecreased so it actually receives events the catalog publishes:

main.py¶

async def main(run_catalog: bool, run_notifications: bool) -> None:
    subscription_store = InMemorySubscriptionStore()
    base_directory = ".mersal-quickstart"

    catalog_app, catalog_activator = make_mersal_app("catalog", base_directory, subscription_store)
    notifications_app, notifications_activator = make_mersal_app("notifications", base_directory, subscription_store)
    accounting_app, accounting_activator = make_mersal_app("accounting", base_directory, subscription_store)

    catalog_activator.register(UpdateProductPrice, lambda _, app: UpdateProductPriceHandler(app))
    notifications_activator.register(PriceDecreased, lambda _, __: NotifyWishlistUsersHandler())
    accounting_activator.register(PriceDecreased, lambda _, __: RecordPriceChangeHandler())
    await notifications_app.subscribe(PriceDecreased)
    await accounting_app.subscribe(PriceDecreased)

    if run_catalog:
        await catalog_app.start()
    if run_notifications:
        await notifications_app.start()
    await accounting_app.start()

    print("Running. Press Ctrl+C to stop.")
    await anyio.sleep_forever()

Adding another consumer¶

Nothing says only one app can react to PriceDecreased. Say the finance team wants to log every price change for reporting, completely independently of notifications. Because pub/sub decouples publishers from subscribers, adding them is just another handler and another subscription - nothing about the catalog or the notifications app has to change.

Create a new file called record_price_change_handler.py:

record_price_change_handler.py¶

from __future__ import annotations

from messages import PriceDecreased

__all__ = ("RecordPriceChangeHandler",)

_price_change_log: list[str] = []

class RecordPriceChangeHandler:
    async def __call__(self, event: PriceDecreased) -> None:
        _price_change_log.append(f"{event.product_id}: ${event.new_price:.2f}")
        print(f"[accounting] Logged price change for {event.product_id}: now ${event.new_price:.2f}")

Create a third Mersal app, register the handler, and subscribe it to PriceDecreased just like we did for notifications:

main.py¶

async def main(run_catalog: bool, run_notifications: bool) -> None:
    subscription_store = InMemorySubscriptionStore()
    base_directory = ".mersal-quickstart"

    catalog_app, catalog_activator = make_mersal_app("catalog", base_directory, subscription_store)
    notifications_app, notifications_activator = make_mersal_app("notifications", base_directory, subscription_store)
    accounting_app, accounting_activator = make_mersal_app("accounting", base_directory, subscription_store)

    catalog_activator.register(UpdateProductPrice, lambda _, app: UpdateProductPriceHandler(app))
    notifications_activator.register(PriceDecreased, lambda _, __: NotifyWishlistUsersHandler())
    accounting_activator.register(PriceDecreased, lambda _, __: RecordPriceChangeHandler())
    await notifications_app.subscribe(PriceDecreased)
    await accounting_app.subscribe(PriceDecreased)

    if run_catalog:
        await catalog_app.start()
    if run_notifications:
        await notifications_app.start()
    await accounting_app.start()

    print("Running. Press Ctrl+C to stop.")
    await anyio.sleep_forever()

That’s the whole change. The catalog still only knows how to publish PriceDecreased - it has no idea it now has two subscribers instead of one.

Wiring it together¶

The catalog, notifications, and accounting apps are three separate Mersal apps, each with its own queue. What lets an event published by one reach subscribers registered on the others is the single subscription_store that make_mersal_app was handed for all three - no extra wiring needed beyond passing the same store to every app.

Sending the first message¶

All three apps know how to handle their messages now. Something still has to put an UpdateProductPrice command on the catalog’s queue in the first place. In a real system, that trigger usually comes from one of:

A legacy or external system - some other piece of infrastructure already writes messages compatible with the transport directly onto the queue, without going through Mersal at all.
Manually placing one - handy while developing or debugging. Our transport is just JSON files on disk, so dropping a file like this into .mersal-quickstart/catalog/ is enough for it to get picked up:
.mersal-quickstart/catalog/manual-message.json¶
```
{
  "headers": {"message_id": "11111111-1111-1111-1111-111111111111"},
  "body": "{\"type\": \"UpdateProductPrice\", \"data\": {\"product_id\": \"product-42\", \"new_price\": 799.0}}",
  "body_type": "json"
}
```
Mersal itself - most commonly, some other part of your app calls send_local or send, for example from an HTTP endpoint right after it receives a request to change a product’s price.

For this quick start we’ll take the third route, and we’ll do it from a second, separate script. main.py is a worker: its only job is to host handlers and process whatever shows up on a queue, so it never places anything on one itself. Placing a message is a different concern, so it gets its own file.

Why bother with two files?

Nothing about Mersal requires this split - the same Python process that runs a worker can just as easily call send_local itself, for example from an HTTP handler that also happens to host the catalog worker. We’re using two scripts here only so the quick start makes the boundary obvious: hosting handlers (what start() does) and placing a message on a queue (what send_local/send/publish do) are independent things, and you only ever need the one you’re actually doing.

Create a new file called send_price_update.py:

send_price_update.py¶

from __future__ import annotations

import anyio
from main import make_mersal_app

from messages import UpdateProductPrice

__all__ = ("main",)

async def main() -> None:
    catalog_app, _ = make_mersal_app("catalog", ".mersal-quickstart")
    await catalog_app.send_local(UpdateProductPrice("product-42", 799.0))

if __name__ == "__main__":
    anyio.run(main)

Notice make_mersal_app("catalog", ...) here is the exact same call main.py makes for the catalog app - same name, same transport, same serializer, so it resolves to the same queue on disk. The only difference is that this script never calls catalog_app.start(). Starting is what turns an app into a worker that pulls messages off its queue and dispatches them to handlers; skip it, and the app object can still place a message on a queue just fine, it simply won’t be the one to pick anything up.

Now, a question worth asking before we send anything: what should happen if the notifications app happens to be down, mid-deploy, or just plain buggy at the moment the price changes? Should the catalog wait for it? It shouldn’t have to - and with Mersal it doesn’t.

To show that, main.py accepts --catalog and --notifications flags that control which of those two apps actually starts (accounting always starts - this isn’t about accounting, it’s here to prove the other two don’t need each other). Passing neither flag starts everything, which is the normal case:

main.py¶

if __name__ == "__main__":
    parser = argparse.ArgumentParser()
    parser.add_argument("--catalog", action="store_true")
    parser.add_argument("--notifications", action="store_true")
    args = parser.parse_args()

    run_both = not args.catalog and not args.notifications
    anyio.run(main, args.catalog or run_both, args.notifications or run_both)

A real worker doesn’t do its job once and exit - it keeps its workers running so it can keep handling whatever shows up next:

main.py¶

async def main(run_catalog: bool, run_notifications: bool) -> None:
    subscription_store = InMemorySubscriptionStore()
    base_directory = ".mersal-quickstart"

    catalog_app, catalog_activator = make_mersal_app("catalog", base_directory, subscription_store)
    notifications_app, notifications_activator = make_mersal_app("notifications", base_directory, subscription_store)
    accounting_app, accounting_activator = make_mersal_app("accounting", base_directory, subscription_store)

    catalog_activator.register(UpdateProductPrice, lambda _, app: UpdateProductPriceHandler(app))
    notifications_activator.register(PriceDecreased, lambda _, __: NotifyWishlistUsersHandler())
    accounting_activator.register(PriceDecreased, lambda _, __: RecordPriceChangeHandler())
    await notifications_app.subscribe(PriceDecreased)
    await accounting_app.subscribe(PriceDecreased)

    if run_catalog:
        await catalog_app.start()
    if run_notifications:
        await notifications_app.start()
    await accounting_app.start()

    print("Running. Press Ctrl+C to stop.")
    await anyio.sleep_forever()

Running it¶

Start the worker in one terminal and leave it running:

uv run main.py

Running. Press Ctrl+C to stop.

Then, in a second terminal, send the price update:

uv run send_price_update.py

Back in the first terminal, the worker’s output appears as soon as the message is picked up:

[catalog] product-42 is now $799.00
[accounting] Logged price change for product-42: now $799.00
[notifications] Emailing [email protected]: product-42 dropped to $799.00
[notifications] Emailing [email protected]: product-42 dropped to $799.00

The worker keeps running after that - press Ctrl+C in its terminal when you’re done watching it. send_price_update.py has already exited on its own; it only ever had one message to send.

What you just built¶

A command (UpdateProductPrice) sent straight to one app with send_local.
An event (PriceDecreased) published with publish and delivered to whoever is subscribed, without the publisher knowing who that is.
Three independent Mersal apps, each running their own worker loop, coordinating only through messages - notifications and accounting were added without ever touching the catalog.

What happens when it fails?¶

Real email providers go down. Let’s see what happens when the notifications handler always fails. Add this next to NotifyWishlistUsersHandler in notify_wishlist_users_handler.py:

notify_wishlist_users_handler.py¶

class FlakyNotifyWishlistUsersHandler:
    def __init__(self) -> None:
        self.attempts = 0

    async def __call__(self, event: PriceDecreased) -> None:
        self.attempts += 1
        print(f"[notifications] Attempt {self.attempts}: emailing wishlist for {event.product_id}...")
        raise ConnectionError("SendGrid API is currently unreachable!")

Then, in main.py, register that one instead:

main.py¶

notifications_activator.register(PriceDecreased, lambda _, __: FlakyNotifyWishlistUsersHandler())

Nothing else changes - no retry configuration, no error queue setup. Mersal’s defaults (RetryStrategySettings) already give every app 5 attempts per message before moving it to a queue named "error":

Start the worker, then send the price update from the second terminal just like before:

uv run main.py

uv run send_price_update.py

Back in the worker’s terminal:

[catalog] product-42 is now $799.00
[accounting] Logged price change for product-42: now $799.00
[notifications] Attempt 1: emailing wishlist for product-42...
[notifications] Attempt 2: emailing wishlist for product-42...
[notifications] Attempt 3: emailing wishlist for product-42...
[notifications] Attempt 4: emailing wishlist for product-42...
[notifications] Attempt 5: emailing wishlist for product-42...

Notice that accounting logs the price change just fine - it’s a completely separate subscriber, so notifications failing and retrying has no effect on it whatsoever. Press Ctrl+C to stop the process once the retries are done.

Revert the registration line back to NotifyWishlistUsersHandler before moving on.

Running one app without the other¶

The --catalog and --notifications flags let us simulate the notifications app being down without touching any code. Start only the catalog (and accounting) side:

uv run main.py --catalog

Then, from the second terminal, send the price update as before:

uv run send_price_update.py

Back in the worker’s terminal:

[catalog] product-42 is now $799.00
[accounting] Logged price change for product-42: now $799.00

Notifications never started, so nothing was there to receive the PriceDecreased event - but it wasn’t dropped either. Because notifications_app.subscribe(PriceDecreased) had already registered its queue address with the shared subscription_store, the catalog delivered the event straight onto disk at .mersal-quickstart/notifications/, where it sits waiting. Press Ctrl+C to stop this run.

Now start only notifications:

uv run main.py --notifications

[notifications] Emailing [email protected]: product-42 dropped to $799.00
[notifications] Emailing [email protected]: product-42 dropped to $799.00
Running. Press Ctrl+C to stop.

Accounting has nothing new to log this time - it already picked up and processed its copy of the event during the first run, since it starts unconditionally regardless of which flags are passed. Notifications, on the other hand, picks up the event it had backlogged: no message was lost while it was “down”. This is the same durability at work as the error queue: messages live on the transport, not in memory, so a subscriber can subscribe long before it ever starts, and catch up on whatever accumulated once it does.

Sending before any worker exists¶

Every run so far started main.py first and sent the update second. Take that further: what if the catalog worker has never run at all - not down mid-deploy, not restarting, just never started once - when the price update needs to go out?

Send the update with nothing running:

uv run send_price_update.py

Nothing prints, and there’s no error. The command connected to the catalog’s queue on disk, wrote UpdateProductPrice into it, and exited - it never needed a worker to be listening.

Now start the worker, whenever that happens to be:

uv run main.py

[catalog] product-42 is now $799.00
[accounting] Logged price change for product-42: now $799.00
[notifications] Emailing [email protected]: product-42 dropped to $799.00
[notifications] Emailing [email protected]: product-42 dropped to $799.00
Running. Press Ctrl+C to stop.

The instant the catalog worker comes online it finds the message that was already waiting, and the whole chain fires from there: the price update gets handled, PriceDecreased gets published, and both accounting and notifications react - despite the fact that none of these three apps existed as a running process at the moment the message was sent. The queue is a directory on disk, not an in-memory mailbox held by a running app, so there’s nothing for a sender to wait on and no ordering requirement between “a worker exists” and “a message arrives for it.”

Press Ctrl+C to stop the worker once you’re done watching it.

Next steps¶

Pub/Sub - subscriptions, topics, and autosubscribe
Error Handling - customizing retries, fail-fast exceptions, and dead-letter handling
Usage - transports, unit of work, sagas, and everything else