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:
        if self._conn is not None:
            return
        bus = await MessageBus().connect()
        try:
            introspection = await bus.introspect(BUS_NAME, OBJECT_PATH)
            proxy = bus.get_proxy_object(BUS_NAME, OBJECT_PATH, introspection)
            iface = proxy.get_interface(INTERFACE_NAME)
            if hasattr(iface, "on_action_invoked"):
                iface.on_action_invoked(self._handle_action)
            if hasattr(iface, "on_notification_closed"):
                iface.on_notification_closed(self._handle_closed)
        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)

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()
    assert self._conn is not None  # connect() post-condition

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

    return await self._conn.interface.call_notify(
        self._app_name,
        replaces_id,
        app_icon or _DEFAULT_APP_ICON,
        summary,
        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
    if hasattr(conn.interface, "off_action_invoked"):
        conn.interface.off_action_invoked(self._handle_action)
    if hasattr(conn.interface, "off_notification_closed"):
        conn.interface.off_notification_closed(self._handle_closed)
    conn.bus.disconnect()
    self._conn = None
    self._callbacks.clear()