notificationcenter

This is a Python port of NotificationCenter from macOS. It is a faithful reimplementation of Apple's NSNotificationCenter (the Foundation/Cocoa class, exposed in Swift simply as NotificationCenter) for Python: a tiny, dependency-free, thread-safe, singleton one-to-many notification bus. One object announces that something happened and any number of others react, without either side knowing about the other. The one deliberate improvement over the original is that notification names must be Enum members rather than strings, so they are typo-proof, autocompletable, and self-documenting.

pip install notifcenter        # distribution name

import notificationcenter          # import name

The plain notificationcenter name on PyPI belongs to an abandoned 2016 package, so the distribution ships as notifcenter while the import stays notificationcenter.

Requirements: Python 3.6+ and zero third-party dependencies (only the standard-library enum and threading). The code itself is even 3.5-clean; 3.6 is the declared floor purely because older tooling is impractical.

Why

Use it when one object needs to tell many others that something happened, without knowing who they are or what they will do. The poster and the observers stay fully decoupled: the poster just announces, and observers react. This is exactly the role NSNotificationCenter plays on macOS and iOS, brought over to Python.

Quick start

from enum import Enum
from notificationcenter import NotificationCenter

class DeviceNotification(Enum):
    will_move = "will_move"
    did_move  = "did_move"

class Logger:
    def __init__(self):
        NotificationCenter().add_observer(self, self.on_move, DeviceNotification.did_move)

    def on_move(self, notification):
        print("moved to", notification.user_info["position"])

logger = Logger()

# somewhere else, with no reference to logger:
NotificationCenter().post_notification(
    DeviceNotification.did_move, notifying_object=some_device,
    user_info={"position": (1, 2, 3)},
)
# -> moved to (1, 2, 3)

API

NotificationCenter() always returns the same singleton instance.

Method	Purpose
add_observer(observer, method, notification_name=None, observed_object=None)	Register method to be called with a Notification when notification_name is posted. If observed_object is given, only notifications from that specific object are forwarded.
remove_observer(observer, notification_name=None, observed_object=None)	Stop notifying observer. Omit notification_name to remove it from every notification.
post_notification(notification_name, notifying_object, user_info=None)	Announce notification_name, invoking every matching observer's callback.
observers_count()	Total number of registered observer entries.
clear()	Remove all observers.

Each callback receives a Notification with these attributes: .name is the Enum member that was posted, .object is the object that posted it (the notifying_object), and .user_info is the optional dict of extra data.

Notification names must be Enum members; passing a string raises ValueError.

Common patterns

Observe only one sender. Pass observed_object to hear a notification only when it comes from a specific object, ignoring the same notification from others:

NotificationCenter().add_observer(
    self, self.on_move, DeviceNotification.did_move, observed_object=my_stage
)

Clean up on teardown. An observer that outlives its usefulness keeps getting called (and keeps the object alive). Remove it when you are done, which is the most common observer-pattern bug:

def close(self):
    NotificationCenter().remove_observer(self)   # unregister from every notification

Update a GUI safely. Callbacks run on the thread that posted the notification, which may be a worker thread. Most GUI toolkits, Tkinter included, are not thread-safe, and you must not call them from another thread, not even root.after, which is itself a Tk call. The portable way is to hand the data to the main thread through a thread-safe queue.Queue that the main thread drains on a timer:

import queue

class PositionLabel:
    def __init__(self, root, label):
        self.root, self.label = root, label
        self._inbox = queue.Queue()
        NotificationCenter().add_observer(self, self.on_move, DeviceNotification.did_move)
        self.root.after(100, self._drain)          # polling loop, started on the main thread

    def on_move(self, notification):               # may run on a worker thread
        self._inbox.put(notification.user_info)    # Queue is thread-safe; no Tk call here

    def _drain(self):                              # always runs on the main thread
        try:
            while True:
                self.label.config(text=str(self._inbox.get_nowait()))
        except queue.Empty:
            pass
        self.root.after(100, self._drain)

Some toolkits do provide a genuinely thread-safe cross-thread post (wxPython's wx.CallAfter, Qt's queued signals) and let you skip the queue; Tkinter is the outlier that does not.

Thread safety

All operations are guarded by a re-entrant lock, so you can add and remove observers and post notifications from multiple threads. Observer callbacks, however, run synchronously on the thread that posted the notification, and they run while that lock is held, so keep them quick and non-blocking. If a callback needs to update a GUI, do not call the toolkit from that thread; use the queue-and-drain pattern shown under "Update a GUI safely" above.

Testing

100% line and branch coverage (17 tests, including doctests in the module). The CI enforces it, so coverage falling under 100% breaks the build.

pip install -e ".[dev]"
coverage run -m pytest -q --doctest-modules src tests
coverage report -m        # fails if coverage drops below 100%

License

MIT © Daniel C. Côté