hbutils.design.observer

Observer pattern implementation for event-driven subscriptions.

This module implements a lightweight observer pattern utility centered on the Observable class, enabling objects (subscribers) to register callbacks for specific events and receive notifications when those events are dispatched.

The module contains the following public components:

• Observable - Event dispatcher supporting subscriptions and notifications.

Note

Events can be defined as an enum.Enum subclass or as an iterable of event identifiers (e.g., strings, integers, or enum members).

Example:

>>> from enum import Enum, unique
>>> from hbutils.design.observer import Observable
>>>
>>> @unique
... class Event(Enum):
...     READY = 'ready'
...     DONE = 'done'
>>>
>>> observable = Observable(Event)
>>> received = []
>>> observable.subscribe(Event.READY, received, 'append')
>>> observable.dispatch(Event.READY)
>>> received
[<Event.READY: 'ready'>]

__all__

hbutils.design.observer.__all__ = ['Observable']

Built-in mutable sequence.

If no argument is given, the constructor creates a new empty list. The argument must be an iterable if specified.

Observable

class hbutils.design.observer.Observable(events: Type[Enum] | list | tuple)

Observable object implementing the observer pattern.

The observable holds a set of events, and subscribers may register callbacks for each event. When an event is dispatched, all associated callbacks are invoked with the event and the observable instance as arguments.

• subscribe() can be used for subscribing to a specific event.

• unsubscribe() can be used for unsubscribing from a specific event.

• dispatch() can be used for broadcasting a specific event, and all the subscribed callbacks will be triggered.

Examples::

>>> from enum import IntEnum, unique
>>> from hbutils.design import Observable
>>>
>>> @unique
... class MyIntEnum(IntEnum):
...     A = 1
...     B = 2
>>>
>>> o = Observable(MyIntEnum)
>>> list_a, list_b = [], []
>>> o.subscribe(MyIntEnum.A, list_a, 'append')  # use list_a.append
>>> o.subscribe(MyIntEnum.B, list_a, lambda v: list_a.append(v))  # custom function.
>>> o.subscribe(MyIntEnum.A, list_b, 'append')  # use list_b.append
>>>
>>> list_a, list_b
([], [])
>>> o.dispatch(MyIntEnum.A)
>>> list_a, list_b
([<MyIntEnum.A: 1>], [<MyIntEnum.A: 1>])
>>> o.dispatch(MyIntEnum.B)
>>> list_a, list_b
([<MyIntEnum.A: 1>, <MyIntEnum.B: 2>], [<MyIntEnum.A: 1>])
>>>
>>> o.unsubscribe(MyIntEnum.A, list_a)
>>> o.dispatch(MyIntEnum.A)
>>> list_a, list_b
([<MyIntEnum.A: 1>, <MyIntEnum.B: 2>], [<MyIntEnum.A: 1>, <MyIntEnum.A: 1>])
>>> o.dispatch(MyIntEnum.B)
>>> list_a, list_b
([<MyIntEnum.A: 1>, <MyIntEnum.B: 2>, <MyIntEnum.B: 2>], [<MyIntEnum.A: 1>, <MyIntEnum.A: 1>])

__init__(events: Type[Enum] | list | tuple) → None

Construct an Observable.

Parameters:

events (_EventSetType) – Set of events, can be a list, tuple, or an enum class.

Note

When an enum class is used, its members will be used as events. For example:

>>> from enum import IntEnum
>>> from hbutils.design import Observable
>>>
>>> class MyIntEnum(IntEnum):
...     A = 1
...     B = 2
>>>
>>> # equals to `Observable([MyIntEnum.A, MyIntEnum.B])`
... o = Observable(MyIntEnum)
>>> o._events  # just for explanation, do not do this on actual use
{<MyIntEnum.A: 1>: {}, <MyIntEnum.B: 2>: {}}

dispatch(event: _EventType) → None

Dispatch an event to all subscribers.

This method triggers all callbacks subscribed to the given event, passing the event and the observable instance as arguments.

Parameters:

event (_EventType) – Event to be dispatched.

subscribe(event: _EventType, subscriber: _SubscriberType, callback: _CallbackType | str | None = None) → None

Subscribe to the given event.

Parameters:

• event (_EventType) – Event to be subscribed.

• subscriber (_SubscriberType) – Subscriber of this subscription.

• callback (Union[_CallbackType, str, None]) – Callback function. If str is given, the method with this name on subscriber will be used. Default is None which means the update method on subscriber will be used.

Raises:

TypeError – If the callback is not callable.

Note

Callback functions should accept no more than 2 positional arguments. For example:

>>> o.subscribe(MyIntEnum.A, 'user1', lambda: 2)  # ok
>>> o.subscribe(MyIntEnum.A, 'user2', lambda event: 2)  # ok
>>> o.subscribe(MyIntEnum.A, 'user3', lambda event, obs: 2)  # ok
>>> o.subscribe(MyIntEnum.A, 'user4', lambda x, y, z: 2)  # invalid

subscribers(event: _EventType) → List[_SubscriberType]

Get subscribers of the given event.

Parameters:

event (_EventType) – Event for querying.

Returns:

A list of subscribers.

Return type:

List[_SubscriberType]

subscriptions(event: _EventType) → List[Tuple[_SubscriberType, _CallbackType]]

Get subscriptions of the given event.

Parameters:

event (_EventType) – Event for querying.

Returns:

A list of tuples with subscribers and their callbacks.

Return type:

List[Tuple[_SubscriberType, _CallbackType]]

unsubscribe(event: _EventType, subscriber: _SubscriberType) → None

Unsubscribe from the given event.

Parameters:

• event (_EventType) – Event to be unsubscribed.

• subscriber (_SubscriberType) – Subscriber of this unsubscription.

Raises:

KeyError – If the subscriber is not found for the given event.