util.observer module¶

An “extended” implementation of the observer/subject design pattern. Instead to the traditional subject calling notify() for every change (which may result in a lot of unnecessary updates happening), we use the “with” clause to more elegantly manage updates:

Within the the Subject’s code…¶

with Subject.ChangeMgr(self) as changeMgr:
  try:
    # make a change
    ...
    changeMgr.registerTopics(("topic1", (value [,...])), ("topic2", (value [,...])))
  except:
    changeMgr.cancel()

In the event the changes are nested (ie: the “<make a change>” section calls other methods with updates), this results in saving up all the notifies until the outer update completes. The registerTopics() method assures that the topics are passed in the eventual notify() method. The cancel() method will suppress the final notify() if all nested updates also resulted in a cancel().

There are two ways you can get event notifications from a Subject object. The first is an Observer object must registers with the Subject as an Observer which obligates the Observer to provide an update() method:

Within the the Observer’s code…¶

  subject.attach(self, "topic1", "topic2")

...

def update(self, subject: Subject, *topics:Tuple[str, Tuple[Any, ...]]) -> None:
  topic:str
  params:tuple[Any, ...]
  for topic, params in topics:
    if topic == "topic1":
      # deal with topic1 and it's params
      ...
    elif topic == "topic2":
      ...

The second way to get a notification is through a callback:

No need for an Observer class…¶

subject.attachCallback(lambda topic, args: print(f'{topic}: {args}), \
                       "topic1", 
                       "topic2")

In this case, the lambda function will be invoked whenever the topic event happens. It could also have been an independent function or a method call.

There is an alternate version of notify(), notifyLater() that works similarly, but deferred until the current call stack unwinds (e.g. until a constructor finishes) instead of firing observers synchronously, mid-construction. Use this anywhere notify() would otherwise let an Observer inspect a not-yet-fully-built object.

Updates (The arguments to Subject.notify() and Observer.update() are expected to be tuples with the first element being a “topic” of str type and the second element being a tuple “arguments” to the topic.

Created on Sep 17, 2024

Code author: Rob Kremer <robkremer@me.com>

util.observer.enabled: bool = True¶: Use the enabled flag if you must disable all notifications, for example, during initialization or reading in new data from a file. Don’t forget to restore it to True.

util.observer.alwaysLater: bool = False¶: Set this to True for force (application wide) ALL calls to notify to behave as calls to notifyLater.

class util.observer.Subject¶

Bases: object

The Subject interface declares a set of methods for managing subscribers.

class ChangeMgr(owner: Subject, later: bool = False)¶

Bases: object

A context manager that collects the topics registered during a change (see registerTopics()) and notifies the subject’s observers once, when the outermost change ends – unless it was cancelled.

__init__(owner: Subject, later: bool = False)¶

Parameters:

owner (Subject) – The subject doing the calling.
later (bool) – Iff this is True, use notifyLater() instead of notify(). Default is False.

owner: Subject¶

cancelled: bool¶

later: bool¶

cancel()¶

registerTopics(topics: list[tuple[str, tuple[Any, ...]]] | tuple[str, tuple[Any, ...]])¶

Parameters:: topics (list[tuple[str, tuple[Any, ...]]] | tuple[str, tuple[Any, ...]])

__init__()¶

_topics: list[tuple[str, tuple[Any, ...]]]¶

_changeDepth: int¶

_confirmedCount: int¶

_observers: list[tuple[Observer, set[str]]]¶

_callbacks: list[tuple[Callable, set[str], int]]¶

attach(observer: Observer, *topics: str) → None¶

Parameters:

observer (Observer)
topics (str)

detach(observer: Observer) → int¶

Parameters:: observer – The observer object to stop send updates to.
Returns:: The number of entries removed (there may be more than 1 if topics were registered on different occasions).

attachCallback(callback: Callable[[str, tuple[Any, ...]], Any], *topics: str) → int¶

Parameters:

callback (Callable[[str, tuple[Any, ...]],None]) – A function taking a str and a args parameter.
topics (str) – [str] topics for which the callback should be called. If omitted, all topics will be called.

Returns:

An int that can be used to detach the callback.

An implementation of an anonymous observer: The callback is called whenever a topic is matched on an event. Use detachCallback() with the return from this method to detach.

detachCallback(code: int)¶

Parameters:: code – An int (that was returned from attachCallback()) identifying the callback to remove.

notify(*topics: tuple[str, tuple[Any, ...]]) → None¶

Parameters:: topics – list[Tuple[str, Any, …]]: A *list of topicName/info tuple describing the topic and any further info parameters.

Trigger an update in each subscriber iff either it has no specific topics or one or more of it’s topics matches.

Returns:
Return type:: None

_notify(*topics: tuple[str, tuple[Any, ...]], observers: list[tuple[Observer, set[str]]] | None = None, callbacks: list[tuple[Callable, set[str], int]] | None = None) → None¶

Parameters:

topics (list[Tuple[str, Any, ...]]) – A *list of topicName/info tuple describing the topic and any further info parameters.
observers (list[Observer]) – The list of observers to use.
callbacks – list[tuple[Callable, set[str], int]]

Trigger an update in each subscriber iff either it has no specific topics or one or more of it’s topics matches.

NOTE: the observers and callbacks seem strange given that the object has them already. But if this method is called “later” (usually on the tk event loop), the observers and callbacks should be those AT THE TIME OF THE ORIGINAL NOTIFICATION, not at the time the notification actually eventually happens.

scheduler: Callable[[...], None] | None = None¶

notifyLater(*topics: tuple[str, tuple[Any, ...]]) → None¶

Like notify(), but deferred until the current call stack unwinds (e.g. until a constructor finishes) instead of firing observers synchronously, mid-construction. Use this anywhere notify() would otherwise let an Observer inspect a not-yet-fully-built object.

Tries, in order: an app-supplied scheduler (self.scheduler / Subject.scheduler); tkinter’s default root’s after_idle(), if tkinter is already imported and has a live root; an asyncio event loop’s call_soon()/call_soon_threadsafe(), if one is running; and finally a threading.Timer(0, …) as a last resort.

self._observers/self._callbacks are snapshotted right here, before any of those schedulers run – not read again when the deferred call actually fires – since by then they may have changed (an Observer attached/detached in the meantime, say), and the whole point of deferring is to still deliver the notification this call was actually asked to send. None of the four scheduling mechanisms below reliably accepts extra keyword arguments to forward to the deferred call (Tk’s after_idle() and asyncio’s call_soon()/call_soon_threadsafe() only take *args at all; threading.Timer does take a kwargs= param, but there’s no reason to special-case it), so the snapshot is captured as plain local variables and closed over by a single, argument-less _fireNow(), which every scheduler is just asked to call once, later, with nothing further to pass through.

Parameters:: topics – Same as notify().
Return type:: None

class util.observer.Observer¶

Bases: ABC

The Observer interface declares the update method, used by subjects.

_abc_impl = <_abc._abc_data object>¶

abstractmethod update(subject: Subject, *topics: tuple[str, tuple[Any, ...]]) → None¶

Receive update from subject.

Parameters:

subject (Subject)
topics (tuple[str, tuple[Any, ...]])