Notifications

Notifications notify zero or more handlers that something has happened. The publisher receives no response and does not select a handler.

Requests and notifications have different contracts

Use a request when the caller needs one operation to produce a result. Use a notification when the publisher only needs to report a completed fact, such as OrderPlaced, and subscribers can act independently.

Define and publish a notification

A notification inherits from Notification. Its handlers inherit from NotificationHandler[EventType], accept that exact notification type, and return None:

import asyncio
from dataclasses import dataclass

from pymediate import Notification, NotificationHandler, Mediator, Services


@dataclass(frozen=True)
class OrderPlaced(Notification):
    order_id: int
    item: str


class RecordOrder(NotificationHandler[OrderPlaced]):
    async def __call__(self, notification: OrderPlaced) -> None:
        print(f"recorded {notification.order_id}")


class NotifyWarehouse(NotificationHandler[OrderPlaced]):
    async def __call__(self, notification: OrderPlaced) -> None:
        await asyncio.sleep(0)
        print(f"warehouse notified about {notification.item}")


async def main() -> None:
    services = Services(RecordOrder(), NotifyWarehouse())
    mediator = Mediator(services)

    await mediator.publish(OrderPlaced(order_id=42, item="tea"))


asyncio.run(main())

Defining each handler class registers its subscription. Passing a handler instance to Services makes that instance available when a notification is published, so every subscriber a published notification needs must be in the collection when it is constructed.

The asynchronous and synchronous APIs share this process-wide subscription list. All handlers for one exact notification type must therefore use the same form: either async def handlers from pymediate.NotificationHandler or def handlers from pymediate.sync.NotificationHandler. Mixing the forms for one notification gives a mediator subscribers it cannot invoke correctly. Use distinct notification types or separate processes when both forms are required.

Dispatch uses the notification instance's exact class. A handler for a base notification class does not receive instances of a derived notification class. The parameter annotation must also match the notification type in NotificationHandler[...] exactly. See the NotificationHandler validation rules.

Delivery and failures

The asynchronous mediator resolves all subscriber instances before any handler runs. It then runs the handlers concurrently with asyncio.gather(). Task creation follows registration order, but completion order depends on the handlers.

The synchronous mediator runs handlers one at a time in registration order. Resolution failures behave the same in both versions, but subscriber failures differ slightly:

• If a subscriber class has no instance in the service provider, ServiceNotFoundError is raised before any handler runs.
• The asynchronous mediator lets every subscriber finish after ordinary Exception failures and raises them in an ExceptionGroup. Other collected BaseException values produce a BaseExceptionGroup. KeyboardInterrupt and SystemExit propagate instead and can cancel unfinished subscribers.
• The synchronous mediator catches ordinary Exception failures, continues through the remaining subscribers, and raises those failures in an ExceptionGroup. A direct BaseException, such as KeyboardInterrupt, stops synchronous delivery and propagates immediately.

Python's except* syntax can handle selected exception types from either group. It does not turn a directly propagated KeyboardInterrupt or SystemExit into a group.

Handlers for one notification should therefore not depend on each other's completion order. Put ordered work in one handler, or model the dependent operation as a request.

Notifications are in-process dispatch

publish() calls handlers in the current process and does not return until they finish. It does not persist notifications, retry failed delivery, or send work to another process. Use a message broker or durable job system when those properties are required. A notification handler can publish to such a system as part of an application's integration code.

Definition-time validation

PyMediate validates a notification handler when Python defines its class. It requires:

• a notification type that inherits from Notification;
• an exact parameter annotation;
• a None return annotation; and
• async def for pymediate.NotificationHandler, or def for pymediate.sync.NotificationHandler.

The corresponding errors are InvalidNotificationTypeError and InvalidHandlerSignatureError. Pipeline behaviors wrap send() only, so they do not run for published notifications.

Testing

Call a notification handler directly when testing its own behavior. Use publish() in the smaller set of tests that need to verify subscriber wiring, concurrency, or grouped failures. Notification subscriptions are registered at class definition and remain process-wide, so define shared test notification and handler classes once. The testing guide describes this constraint in more detail.

Continue

• Requests and responses covers single-response dispatch.
• Streaming covers one handler yielding several typed chunks.
• Error handling lists PyMediate's exceptions.