desktop

desktop

Desktop notifier backed by dbus-fast and the freedesktop Notifications spec.

BUS_NAME = 'org.freedesktop.Notifications' module-attribute

OBJECT_PATH = '/org/freedesktop/Notifications' module-attribute

INTERFACE_NAME = 'org.freedesktop.Notifications' module-attribute

CloseReason

Bases: IntEnum

Reason a notification was closed, per the freedesktop spec.

EXPIRED = 1 class-attribute instance-attribute

The notification expired (timed out).

DISMISSED = 2 class-attribute instance-attribute

The notification was dismissed by the user.

CLOSED = 3 class-attribute instance-attribute

The notification was closed via CloseNotification.

UNDEFINED = 4 class-attribute instance-attribute

The notification server did not provide a reason.

DbusNotifier(app_name='terok')

Send desktop notifications over the D-Bus session bus.

The connection is established lazily on the first notify call. Action callbacks are dispatched from the ActionInvoked signal; stale callbacks are cleaned up automatically on NotificationClosed.

Parameters:

Name	Type	Description	Default
app_name	str	Application name sent with every notification.	'terok'

Initialise with the given application name.

Source code in src/terok_clearance/notifications/desktop.py

def __init__(self, app_name: str = "terok") -> None:
    """Initialise with the given application name."""
    self._app_name = app_name
    self._conn: _Connection | None = None
    self._callbacks: dict[int, Callable[[str], None]] = {}
    self._connect_lock = asyncio.Lock()

connect() async

Idempotently open the session-bus connection and subscribe to signals.

Safe to call concurrently and repeatedly: the lock serialises racing callers so exactly one MessageBus is ever created for this notifier.

Source code in src/terok_clearance/notifications/desktop.py

async def connect(self) -> None:
    """Idempotently open the session-bus connection and subscribe to signals.

    Safe to call concurrently and repeatedly: the lock serialises racing
    callers so exactly one MessageBus is ever created for this notifier.
    """
    if self._conn is not None:
        return
    async with self._connect_lock:
        # Double-checked locking: another task may have set ``_conn``
        # between the first check and acquiring the lock.  Mypy can't
        # see the concurrent write so it treats the second check as
        # unreachable.
        if self._conn is not None:
            return  # type: ignore[unreachable]
        # AUTH EXTERNAL: the daemon authenticates by SO_PEERCRED, so the
        # advertised UID must be the connecting process's UID *as the
        # daemon's user namespace sees it*.  Reaching the operator's host
        # session bus from inside a rootless container, that is the outer
        # (host) UID (``host_uid()``); for a bus in our own namespace (an
        # in-namespace sidecar, or a dbusmock test daemon) it is the inner
        # ``os.geteuid()``.  Try the outer UID first (the desktop-notify
        # case) and fall back to the inner one, so both topologies work.
        # When there is no userns translation the two are equal and this
        # is a single attempt — identical to the old behaviour.
        candidate_uids: list[int] = []
        for _uid in (host_uid(), os.geteuid()):
            if _uid not in candidate_uids:
                candidate_uids.append(_uid)
        bus = None
        last_exc: Exception | None = None
        for _uid in candidate_uids:
            try:
                bus = await MessageBus(auth=AuthExternal(uid=_uid)).connect()
                break
            except Exception as exc:  # noqa: BLE001 — retry with next UID
                last_exc = exc
        if bus is None:
            # candidate_uids is non-empty, so the loop ran and set last_exc.
            raise cast(Exception, last_exc)
        try:
            # Build the proxy from a hand-rolled XML — the
            # spec-defined shape — instead of a runtime introspect
            # so the method-call surface (``call_notify``,
            # ``call_close_notification``) is available without a
            # round-trip and without depending on what the session
            # daemon's Introspect happens to return.
            proxy = bus.get_proxy_object(
                BUS_NAME,
                OBJECT_PATH,
                _IntrospectionNode.parse(_NOTIFICATIONS_INTROSPECTION_XML),
            )
            iface = proxy.get_interface(INTERFACE_NAME)
            # Subscribe via a raw bus message handler instead of
            # the proxy's ``on_<signal>`` setattrs.  dbus_fast's
            # proxy interface filters incoming signals on a
            # ``msg.sender == bus._name_owners[bus_name]`` check
            # — which silently drops every ``ActionInvoked`` on
            # relay-fronted setups (xdg-desktop-portal in front
            # of GNOME Shell, for one) where the well-known
            # ``org.freedesktop.Notifications`` is owned by the
            # relay but signals arrive from whichever process
            # the relay forwards through.  The send path
            # (``call_notify``) is unaffected by that filter,
            # which is exactly the asymmetric symptom that
            # produced this fix: popups appear, button clicks
            # vanish.  A raw handler + a sender-agnostic match
            # rule ([`_SIGNAL_MATCH_RULE`][terok_clearance.notifications.desktop.DbusNotifier._SIGNAL_MATCH_RULE])
            # bypasses ``_name_owners`` entirely.
            bus.add_message_handler(self._dispatch_signal)
            # Diagnostic: confirm the handler actually landed in the
            # bus's handler list.  ``getattr`` defends against future
            # dbus_fast renames of the private cache.
            handler_count = len(getattr(bus, "_user_message_handlers", []))
            _log.info(
                "registered _dispatch_signal — bus has %d user handler(s)",
                handler_count,
            )
            await self._add_signal_match(bus)
        except BaseException:
            # Catch ``BaseException`` so an ``asyncio.CancelledError``
            # (``BaseException`` subclass on 3.11+) mid-handshake doesn't
            # leak the already-connected bus.
            bus.disconnect()
            raise
        self._conn = _Connection(bus=bus, interface=iface)
        _log.info(
            "DbusNotifier connected as %r — ActionInvoked / NotificationClosed subscribed",
            self._app_name,
        )

notify(summary, body='', *, actions=(), timeout_ms=-1, hints=None, replaces_id=0, app_icon='', container_id='', container_name='', project='', task_id='', task_name='') async

Send a desktop notification.

Freedesktop notifications render summary + body + actions only, so the structured identity kwargs (container_id and the terok task triple) are dropped on the floor here — callers are expected to have folded the user-facing identity into body already. The kwargs stay in the signature for Notifier conformance so callers don't have to branch on notifier kind.

Source code in src/terok_clearance/notifications/desktop.py

async def notify(
    self,
    summary: str,
    body: str = "",
    *,
    actions: Sequence[tuple[str, str]] = (),
    timeout_ms: int = -1,
    hints: Mapping[str, Any] | None = None,
    replaces_id: int = 0,
    app_icon: str = "",
    container_id: str = "",  # noqa: ARG002 — protocol kwarg ignored by desktop
    container_name: str = "",  # noqa: ARG002 — protocol kwarg ignored by desktop
    project: str = "",  # noqa: ARG002 — protocol kwarg ignored by desktop
    task_id: str = "",  # noqa: ARG002 — protocol kwarg ignored by desktop
    task_name: str = "",  # noqa: ARG002 — protocol kwarg ignored by desktop
) -> int:
    """Send a desktop notification.

    Freedesktop notifications render summary + body + actions only,
    so the structured identity kwargs (``container_id`` and the
    terok task triple) are dropped on the floor here — callers are
    expected to have folded the user-facing identity into ``body``
    already.  The kwargs stay in the signature for
    [`Notifier`][terok_clearance.notifications.protocol.Notifier] conformance so callers
    don't have to branch on notifier kind.
    """
    await self.connect()
    conn = cast(_Connection, self._conn)  # connect() post-condition

    actions_flat: list[str] = []
    for action_id, label in actions:
        actions_flat.extend((action_id, label))

    return await conn.interface.call_notify(
        self._app_name,
        replaces_id,
        app_icon or _default_app_icon(),
        _pango_escape(summary),
        _pango_escape(body),
        actions_flat,
        dict(hints) if hints is not None else {},
        timeout_ms,
    )

on_action(notification_id, callback) async

Register a callback for when the user clicks an action button.

Parameters:

Name	Type	Description	Default
notification_id	int	ID returned by notify.	required
callback	Callable[[str], None]	Called with the action_id string when invoked.	required

Source code in src/terok_clearance/notifications/desktop.py

async def on_action(
    self,
    notification_id: int,
    callback: Callable[[str], None],
) -> None:
    """Register a callback for when the user clicks an action button.

    Args:
        notification_id: ID returned by ``notify``.
        callback: Called with the ``action_id`` string when invoked.
    """
    self._callbacks[notification_id] = callback

close(notification_id) async

Close an active notification.

Parameters:

Name	Type	Description	Default
notification_id	int	ID returned by notify.	required

Source code in src/terok_clearance/notifications/desktop.py

async def close(self, notification_id: int) -> None:
    """Close an active notification.

    Args:
        notification_id: ID returned by ``notify``.
    """
    self._callbacks.pop(notification_id, None)
    if self._conn is not None:
        await self._conn.interface.call_close_notification(notification_id)

disconnect() async

Tear down the session-bus connection.

Source code in src/terok_clearance/notifications/desktop.py

async def disconnect(self) -> None:
    """Tear down the session-bus connection."""
    conn = self._conn
    if conn is None:
        return
    try:
        conn.bus.remove_message_handler(self._dispatch_signal)
    except Exception:
        # Best-effort: a torn-down bus already dropped its handler
        # registry, and we're disconnecting anyway.
        _log.debug("remove_message_handler raised during disconnect", exc_info=True)
    conn.bus.disconnect()
    self._conn = None
    self._callbacks.clear()