Skip to content

_notifier

_notifier

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

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_dbus/_notifier.py
def __init__(self, app_name: str = "terok") -> None:
    """Initialise with the given application name."""
    self._app_name = app_name
    self._bus: MessageBus | None = None
    self._interface: object | None = None
    self._callbacks: dict[int, Callable[[str], None]] = {}
    self._connect_lock = asyncio.Lock()

notify(summary, body='', *, actions=(), timeout_ms=-1, hints=None, replaces_id=0, app_icon='') async

Send a desktop notification.

Parameters:

Name Type Description Default
summary str

Notification title.

required
body str

Optional body text.

''
actions Sequence[tuple[str, str]]

(action_id, label) pairs rendered as buttons.

()
timeout_ms int

Expiration hint in milliseconds (-1 = server default).

-1
hints Mapping[str, Any] | None

Freedesktop hint dict (values should be dbus_fast.Variant).

None
replaces_id int

Replace an existing notification in-place.

0
app_icon str

Icon name or file:/// URI.

''

Returns:

Type Description
int

Server-assigned notification ID.

Source code in src/terok_dbus/_notifier.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 = "",
) -> int:
    """Send a desktop notification.

    Args:
        summary: Notification title.
        body: Optional body text.
        actions: ``(action_id, label)`` pairs rendered as buttons.
        timeout_ms: Expiration hint in milliseconds (``-1`` = server default).
        hints: Freedesktop hint dict (values should be ``dbus_fast.Variant``).
        replaces_id: Replace an existing notification in-place.
        app_icon: Icon name or ``file:///`` URI.

    Returns:
        Server-assigned notification ID.
    """
    if self._interface is None:
        async with self._connect_lock:
            if self._interface is None:
                await self._connect()

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

    return await self._interface.call_notify(  # type: ignore[union-attr]
        self._app_name,
        replaces_id,
        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_dbus/_notifier.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_dbus/_notifier.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._interface is not None:
        await self._interface.call_close_notification(notification_id)  # type: ignore[union-attr]

disconnect() async

Tear down the session-bus connection.

Source code in src/terok_dbus/_notifier.py
async def disconnect(self) -> None:
    """Tear down the session-bus connection."""
    if self._interface is not None:
        if hasattr(self._interface, "off_action_invoked"):
            self._interface.off_action_invoked(self._handle_action)
        if hasattr(self._interface, "off_notification_closed"):
            self._interface.off_notification_closed(self._handle_closed)
    if self._bus is not None:
        self._bus.disconnect()
    self._bus = None
    self._interface = None
    self._callbacks.clear()