Bridge — API reference¶

Interface¶

BridgeChannel ¶

Bases: ABC

A swappable Bridge transport.

Subclasses declare their identity and config, know how to test themselves, and implement the two transport verbs (:meth:drain inbound, :meth:publish outbound). Register a subclass in :mod:~mait_code.bridge.registry and it becomes selectable everywhere the gate, form and hooks already handle — with no change to those callers.

Class attributes

type_id: Stable, lowercase identifier (the bridge-type value). display_name: Human label for the channel selector. hidden: True keeps the channel out of the user-facing picker while still usable by id (the loopback test double sets this).

config_schema `abstractmethod` `classmethod` ¶

config_schema() -> Sequence[ConfigField]

Return the config inputs this channel needs, in display order.

from_config `abstractmethod` `classmethod` ¶

from_config(config: Mapping[str, str]) -> BridgeChannel

Build a channel from resolved config values.

RAISES	DESCRIPTION
`ValueError`	If a required value is missing or malformed. Callers turn this into a form error or a `doctor` warning, never a crashed session.

test_connection `abstractmethod` ¶

test_connection() -> TestResult

Attempt a lightweight round-trip and report the outcome.

Must never raise: connection/credential failures come back as a TestResult(ok=False, …) so the form can show them inline.

drain `abstractmethod` ¶

drain(since: str | None) -> DrainResult

Fetch inbound captures after the since watermark.

PARAMETER	DESCRIPTION
`since`	The opaque watermark from the previous drain, or `None` on the first drain. TYPE: `str \| None`

publish `abstractmethod` ¶

publish(message: OutboundMessage) -> None

Publish an outbound notification. (Reminders half — #78.)

ConfigField `dataclass` ¶

ConfigField(
    key: str,
    label: str,
    help: str = "",
    kind: str = "str",
    secret: bool = False,
    required: bool = True,
    placeholder: str = "",
)

One user-facing config input a channel needs, declared by the channel.

The Bridge settings form renders one labelled input per field returned by :meth:BridgeChannel.config_schema, so the form is generic over channels rather than hard-coded to any one transport.

ATTRIBUTE	DESCRIPTION
`key`	Stable identifier; the storage key within the channel's config. TYPE: `str`
`label`	Human label shown beside the input. TYPE: `str`
`help`	One-line hint rendered under the input. TYPE: `str`
`kind`	`"str"` \| `"int"` \| `"bool"` — drives input styling/coercion. TYPE: `str`
`secret`	`True` masks the value in displays (tokens, passwords). TYPE: `bool`
`required`	`True` means the channel cannot operate without it; a blank required field is what `doctor` warns about when the gate is on. TYPE: `bool`
`placeholder`	Example value shown in the empty input. TYPE: `str`

TestResult `dataclass` ¶

TestResult(ok: bool, message: str)

Outcome of :meth:BridgeChannel.test_connection — shown inline in the form.

ATTRIBUTE	DESCRIPTION
`ok`	Whether the probe round-trip succeeded. TYPE: `bool`
`message`	Human-readable detail (the error, or a success note). TYPE: `str`

Capture `dataclass` ¶

Capture(body: str, external_id: str = '')

One inbound item drained from a channel, bound for the inbox.

ATTRIBUTE	DESCRIPTION
`body`	The captured text, filed verbatim into `inbox.db`. TYPE: `str`
`external_id`	The channel's own id for the source message, for tracing and de-duplication (empty when the transport has none). TYPE: `str`

DrainResult `dataclass` ¶

DrainResult(
    captures: list[Capture] = list(),
    watermark: str | None = None,
)

What a single :meth:BridgeChannel.drain returned.

The watermark is opaque to the caller: each channel defines what it means (ntfy: the last message id; loopback: a running count). The drain service persists it verbatim per-machine and hands it back on the next drain, so re-drains are idempotent without the service knowing the transport.

ATTRIBUTE	DESCRIPTION
`captures`	Inbound items, oldest first. TYPE: `list[Capture]`
`watermark`	Channel-defined resume token, or `None` to leave the stored watermark unchanged (e.g. nothing new). TYPE: `str \| None`

OutboundMessage `dataclass` ¶

OutboundMessage(
    body: str,
    title: str | None = None,
    actions: tuple[Mapping[str, str], ...] = (),
)

A notification to publish outward (exercised by the reminders half, #78).

Defined here so the interface is complete and both channels implement it now; the reminders publish path that fills in actions lands in a follow-up.

ATTRIBUTE	DESCRIPTION
`body`	The notification text. TYPE: `str`
`title`	Optional title/heading. TYPE: `str \| None`
`actions`	Transport-specific action descriptors (e.g. an ntfy "Done" button); empty for a plain notification. TYPE: `tuple[Mapping[str, str], ...]`

Channels¶

NtfyChannel ¶

NtfyChannel(
    *,
    server: str,
    capture_topic: str,
    notify_topic: str = "",
    token: str = "",
)

Bases: BridgeChannel

Publish/drain a private ntfy topic over HTTP.

test_connection ¶

test_connection() -> TestResult

Poll the topic with a tiny window to prove reachability + auth.

drain ¶

drain(since: str | None) -> DrainResult

Poll messages published since the watermark (a message id, or 'all').

publish ¶

publish(message: OutboundMessage) -> None

POST a notification to the notify topic.

Any control actions become ntfy action buttons that POST the control body back to the capture topic, where the next drain acts on it.

RAISES	DESCRIPTION
`ValueError`	If no notify topic is configured — outbound needs one.

LoopbackChannel ¶

LoopbackChannel(name: str = 'default')

Bases: BridgeChannel

An in-memory transport whose queue is seeded and inspected by tests.

reset `classmethod` ¶

reset() -> None

Forget all queues — call between tests.

seed `classmethod` ¶

seed(*bodies: str, name: str = 'default') -> None

Append inbound messages to a queue, as if published to it.

loop `classmethod` ¶

loop(name: str = 'default') -> _Loop

Return a queue's state for assertions (drain_calls, published, …).

Registry¶

CHANNELS `module-attribute` ¶

CHANNELS: dict[str, type[BridgeChannel]] = {
    (type_id): cls for cls in (NtfyChannel, LoopbackChannel)
}

get_channel_class ¶

get_channel_class(
    type_id: str,
) -> type[BridgeChannel] | None

Return the channel class for type_id, or None if unregistered.

selectable_channels ¶

selectable_channels() -> list[type[BridgeChannel]]

Return the channels offered in the user-facing picker (hidden ones out).

Config & state¶

bridge_enabled ¶

bridge_enabled() -> bool

Whether the Bridge is switched on. Off by default — the safety spine.

active_type ¶

active_type() -> str

The selected channel type id (bridge-type).

active_channel ¶

active_channel() -> BridgeChannel

Build the currently-selected channel from its stored config.

RAISES	DESCRIPTION
`ValueError`	If the type is unknown or its config is incomplete.

build_channel ¶

build_channel(
    type_id: str, values: dict[str, str]
) -> BridgeChannel

Construct a channel from values.

RAISES	DESCRIPTION
`ValueError`	If type_id is unregistered or the config is incomplete.

load_channel_config ¶

load_channel_config(type_id: str) -> dict[str, str]

Return the stored config for one channel type ({} if none).

save_channel_config ¶

save_channel_config(
    type_id: str, values: dict[str, str]
) -> None

Persist config for one channel type, leaving other channels untouched.

get_watermark ¶

get_watermark(type_id: str) -> str | None

Return the last drain watermark for a channel on this machine.

set_watermark ¶

set_watermark(type_id: str, watermark: str) -> None

Record the drain watermark for a channel on this machine.

missing_required ¶

missing_required(
    type_id: str, values: dict[str, str]
) -> list[str]

Return the labels of required config fields left blank.

config_problems ¶

config_problems() -> list[str]

Human-readable configuration problems for doctor (empty when healthy).

Only meaningful when the gate is on: a disabled Bridge needs no config.

Drain & publish service¶

run_drain ¶

run_drain() -> DrainOutcome

Drain the active channel into the inbox, best-effort.

Never raises: every failure mode comes back as a :class:DrainOutcome so a session-start hook can swallow it to a log line and carry on.

drain_channel ¶

drain_channel(
    channel: BridgeChannel, type_id: str
) -> tuple[int, int]

Drain one channel; file captures, act on control messages.

Returns (filed, dismissed). Idempotent across runs: the per-machine watermark is read before and advanced after, so a re-drain from the same position yields nothing new.

DrainOutcome `dataclass` ¶

DrainOutcome(
    status: str,
    count: int = 0,
    dismissed: int = 0,
    detail: str = "",
)

Result of a drain attempt.

ATTRIBUTE	DESCRIPTION
`status`	`"disabled"` (gate off), `"unconfigured"` (gate on but the channel isn't fully set up), `"error"` (the drain raised), or `"ok"`. TYPE: `str`
`count`	Number of captures filed into the inbox. TYPE: `int`
`dismissed`	Number of reminders dismissed by control messages. TYPE: `int`
`detail`	Human-readable context for the non-ok statuses. TYPE: `str`

publish_due_reminders ¶

publish_due_reminders() -> PublishOutcome

Publish due, not-yet-notified reminders outward, best-effort.

Each carries a "Done" action that round-trips a dismissal back through the capture topic. Publishes each reminder once (stamping notified_at), so a still-overdue reminder isn't re-sent every session. Never raises.

PublishOutcome `dataclass` ¶

PublishOutcome(
    status: str, count: int = 0, detail: str = ""
)

Result of an outbound reminder-publish attempt.

ATTRIBUTE	DESCRIPTION
`status`	`"disabled"`, `"unconfigured"`, `"error"`, or `"ok"`. TYPE: `str`
`count`	Number of reminders published outward. TYPE: `int`
`detail`	Human-readable context for the non-ok statuses. TYPE: `str`