hbutils.design.observer

Overview:

Implementation of observer pattern. See Observer Pattern - Wikipedia.

Observable

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

Overview:

Observable object.

• 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: Union[Type[enum.Enum], list, tuple])

Constructor of Observable.

Parameters:

events – 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.

Parameters:

event – Event to be dispatched.

get_subscribers(event: _EventType) → Dict[_SubscriberType, _CallbackType]

Get subscribers of the given event. :param event: Event for querying. :return: A dictionary with subscribers and their callbacks.

subscribe(event: _EventType, subscriber: _SubscriberType, callback: Optional[Union[_CallbackType, str]] = None)

Subscribe to the given event.

Parameters:

• event – Event to be subscribed.

• subscriber – Subscriber of this subscription.

• callback – 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.

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

unsubscribe(event: _EventType, subscriber: _SubscriberType)

Unsubscribe from the given event.

Parameters:

• event – Event to be unsubscribed.

• subscriber – Subscriber of this unsubscription.