Observable / Observer

The Observer pattern defines a one-to-many dependency between objects so that when one object changes state, all its dependents (observers) are notified and updated automatically.

This module provides a flexible implementation of the pattern, supporting both synchronous and asynchronous observers.

Overview

• The Observable class maintains a list of observers.

• Observers can be added using .add_observer() or the += operator.

• Observers are notified via .notify() (non-blocking) or await .notify_async() (fully awaited).

• Observers can be either: - Observer: synchronous, implements .notify(event, data) - AsyncObserver: asynchronous, implements async def notify(event, data)

This design is suitable for both event-driven systems and real-time notification use cases.

Example Usage

from pattern_kit import Observable, Observer, AsyncObserver

class SyncListener(Observer):
    def notify(self, event, data=None):
        print(f"[Sync] Event: {event}, Data: {data}")

class AsyncListener(AsyncObserver):
    async def notify(self, event, data=None):
        print(f"[Async] Event: {event}, Data: {data}")

obs = Observable()
obs += SyncListener()
obs += AsyncListener()

obs.notify("on_event", {"foo": "bar"})         # Non-blocking notify
await obs.notify_async("on_event", {"foo": "bar"})  # Awaited notify_async

API Reference

class pattern_kit.behavioral.observer.Observable

Bases: object

Observable class that supports both synchronous and asynchronous observers.

Observers can be added via add_observer() or using += operator, and removed via remove_observer() or -= operator.

You can trigger notifications using either notify() (non-blocking) or await notify_async() (fully awaited).

add_observer(observer: Observer | AsyncObserver) → None

Add an observer to the list of subscribers.

notify(event: str, data: Any = None) → None

Notify all observers. If an observer is asynchronous, it will be scheduled via asyncio.create_task() (non-blocking).

async notify_async(event: str, data: Any = None) → None

Notify all observers and await any async ones. Sync observers will be called as normal.

remove_observer(observer: Observer | AsyncObserver) → None

Remove an observer from the list of subscribers.

class pattern_kit.behavioral.observer.Observer

Bases: ABC

abstractmethod notify(event: str, data: Any = None) → None

class pattern_kit.behavioral.observer.AsyncObserver

Bases: ABC

abstractmethod async notify(event: str, data: Any = None) → None