hbutils.design.observer

Overview:

Implementation of observer pattern. See Observer Pattern - Wikipedia.

This module provides the Observable class which implements the observer design pattern, allowing objects to subscribe to and receive notifications about events.

Observable

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

Overview:

Observable object implementing the observer pattern.

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

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

• dispatch() can be used for broadcasting a specific event, and all the subscribed callback 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)

Constructor of Observable.

Parameters:

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

Note

When enum is used, its values 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)

Dispatch event to all subscribers.

Parameters:

event (_EventType) – Event to be dispatched.

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

subscribe(event: _EventType, subscriber: _SubscriberType, callback: _CallbackType | str | 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, 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 function should have 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)  # X

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)

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.