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])[source]¶ - 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])[source]¶ 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>: {}}
-
subscribe
(event: _EventType, subscriber: _SubscriberType, callback: Optional[Union[_CallbackType, str]] = None)[source]¶ 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 onsubscriber
will be used. Default isNone
which means theupdate
method onsubscriber
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
-
subscribers
(event: _EventType) → List[_SubscriberType][source]¶ Get subscribers of the given
event
.- Parameters:
event – Event for querying.
- Returns:
A list of subscribers.