correspond.ops

The operations: small protocols an adapter implements a subset of, and the verbs that call them.

An adapter is any object with a name, capabilities and parse_ref(id), plus the methods of whichever protocols it can honour:

Reader	read(ref, *, since=None, limit=None) -> Iterable[Message]
Listener	poll(ref, *, cursor=None, limit=None) -> Iterable[Event]
Writer	send(ref, draft, *, dry_run=False) -> SendResult
Editor	edit(ref, message_id, draft, *, dry_run=False) -> SendResult
Reactor	react(ref, message_id, reaction, *, dry_run=False) -> SendResult
Uploader	upload(ref, name, data, *, media_type, dry_run=False) -> SendResult
Verifier	verify(headers, body) -> Authenticity
AudienceReader	audience(ref, *, draft=None) -> Audience
Labeler	label(ref, labels, *, dry_run=False) -> SendResult
Unlabeler	unlabel(ref, labels, *, dry_run=False) -> SendResult
Closer	close(ref, *, reason=None, dry_run=False) -> SendResult
Reopener	reopen(ref, *, dry_run=False) -> SendResult

The verbs here (read(), listen(), send(), …) take a reference string, find the adapter in the registry, and raise NotSupported naming the operation when the adapter lacks it; never a silent no-op. Writes check the draft against the channel’s capabilities first, and turn a ChannelError into a SendResult with ok=False, so a failed notification never crashes its caller. dry_run=True sends nothing and changes nothing; the only call it makes is the audience lookup. audience() is the exception to refusing: it never raises, because an audience nobody can compute is public.

send() takes an idempotency_key: the same key again posts nothing a second time, even after a send that went out and then reported a failure (correspond.idempotency).

Before every write (send(), edit(), react(), upload()), and in its dry run, the before_send check runs with the conversation’s audience (correspond.outbound). Its verdict and the audience in words go into the plan, and a check that refuses, holds for approval, cannot be loaded or fails stops the write with that error_kind. Calling an adapter’s own methods skips the check: write through these verbs.

Module Attributes

PROTOCOLS

Operation name → the protocol an adapter implements to have it.

Functions

audience(ref[, draft, registry])	Who can read a conversation, asked of its channel now.
capabilities(channel, *[, registry])	What a channel can do, graded, with its limits.
close(ref, *[, reason, dry_run, registry, ...])	Close a conversation (an issue or a discussion), with an optional reason in the channel's vocabulary; the before_send check sees the reason (or close) as the draft's text.
edit(ref, message_id, text, *[, dry_run, ...])	Replace the text of a message correspond's account wrote; the before_send check runs first, as for send().
get_channel(name, *[, registry])	The adapter registered under name; UnknownChannel says what to do if there is none.
implemented(adapter)	The operations an adapter implements, in OPERATIONS order.
label(ref, labels, *[, dry_run, registry, ...])	Add labels to a conversation; the before_send check sees them, comma-joined, as the draft's text.
listen(ref, *[, cursors, limit, commit, ...])	Events since the cursor stored for ref; each cursor is stored once the consumer moves past its event.
parse_ref(ref, *[, registry])	A reference normalised by its channel's adapter (kind and parent filled in, the id validated).
react(ref, message_id, reaction, *[, ...])	Add a reaction to a message (the channel's capabilities list the reactions it accepts); the before_send check sees the reaction as the draft's text.
read(ref, *[, since, limit, registry])	The messages of a conversation, oldest first (limit keeps the most recent ones).
reopen(ref, *[, dry_run, registry, before_send])	Reopen a closed conversation; the before_send check sees reopen as the draft's text.
send(ref, text, *[, title, reply_to, ...])	Send text (or a Draft) to a conversation; dry_run shows the plan and sends nothing.
unlabel(ref, labels, *[, dry_run, registry, ...])	Remove labels from a conversation; the before_send check sees them, comma-joined, as the draft's text.
upload(ref, name, data, *[, media_type, ...])	Send a file to a conversation.
verify(channel, headers, body, *[, registry])	Grade an inbound delivery on channel from its headers and raw body.
webhook_info(channel, *[, registry])	Where channel's platform delivers its updates now (an empty url: polling).
set_webhook(channel, url, *[, dry_run, ...])	Have channel's platform POST its updates to url: listening by polling stops until delete_webhook().
delete_webhook(channel, *[, ...])	Stop webhook delivery for channel, so it can be polled again; the before_send check runs (operation="delete_webhook").
window(messages, *[, since, limit])	Messages sent or edited at or after since, oldest first, keeping the last limit.
with_final_cursor(events, final_cursor)	Yield events, then hand final_cursor to listen(), so a quiet poll still moves the cursor forward.

Classes

AudienceReader(*args, **kwargs)	Who can read a conversation, now and plausibly later, asked of the platform at the time of the call.
Channel(*args, **kwargs)	What every adapter has: a name, its capabilities, and the grammar of its conversation ids.
Closer(*args, **kwargs)	Settle a conversation (close an issue or a discussion), optionally saying why.
Editor(*args, **kwargs)	Replace the text of a message correspond's account wrote.
Labeler(*args, **kwargs)	Add labels to a conversation, apart from its opening send.
Listener(*args, **kwargs)	Events since cursor, oldest first, each carrying the cursor to resume after it.
Reactor(*args, **kwargs)	Add a reaction to a message.
Reader(*args, **kwargs)	Messages of a conversation, oldest first; limit keeps the most recent.
Reopener(*args, **kwargs)	Unsettle a closed conversation.
Unlabeler(*args, **kwargs)	Remove labels from a conversation.
Uploader(*args, **kwargs)	Send a file to a conversation.
Verifier(*args, **kwargs)	Grade an inbound delivery (a webhook, a posted report) from its headers and raw body.
WebhookManager(*args, **kwargs)	Move where a channel's platform delivers its updates.
WebhookReader(*args, **kwargs)	Where a channel's platform delivers its updates now: polling, or a webhook it POSTs to.
Writer(*args, **kwargs)	Send a draft to a conversation.

class correspond.ops.AudienceReader(*args, **kwargs)

Bases: Protocol

Who can read a conversation, now and plausibly later, asked of the platform at the time of the call.

class correspond.ops.Channel(*args, **kwargs)

Bases: Protocol

What every adapter has: a name, its capabilities, and the grammar of its conversation ids.

class correspond.ops.Closer(*args, **kwargs)

Bases: Protocol

Settle a conversation (close an issue or a discussion), optionally saying why.

class correspond.ops.Editor(*args, **kwargs)

Bases: Protocol

Replace the text of a message correspond’s account wrote.

class correspond.ops.Labeler(*args, **kwargs)

Bases: Protocol

Add labels to a conversation, apart from its opening send.

class correspond.ops.Listener(*args, **kwargs)

Bases: Protocol

Events since cursor, oldest first, each carrying the cursor to resume after it.

correspond.ops.PROTOCOLS: dict[str, type] = {'audience': <class 'correspond.ops.AudienceReader'>, 'close': <class 'correspond.ops.Closer'>, 'edit': <class 'correspond.ops.Editor'>, 'label': <class 'correspond.ops.Labeler'>, 'listen': <class 'correspond.ops.Listener'>, 'react': <class 'correspond.ops.Reactor'>, 'read': <class 'correspond.ops.Reader'>, 'reopen': <class 'correspond.ops.Reopener'>, 'send': <class 'correspond.ops.Writer'>, 'unlabel': <class 'correspond.ops.Unlabeler'>, 'upload': <class 'correspond.ops.Uploader'>, 'verify': <class 'correspond.ops.Verifier'>}

Operation name → the protocol an adapter implements to have it.

class correspond.ops.Reactor(*args, **kwargs)

Bases: Protocol

Add a reaction to a message.

class correspond.ops.Reader(*args, **kwargs)

Bases: Protocol

Messages of a conversation, oldest first; limit keeps the most recent.

class correspond.ops.Reopener(*args, **kwargs)

Bases: Protocol

Unsettle a closed conversation.

class correspond.ops.Unlabeler(*args, **kwargs)

Bases: Protocol

Remove labels from a conversation.

class correspond.ops.Uploader(*args, **kwargs)

Bases: Protocol

Send a file to a conversation.

class correspond.ops.Verifier(*args, **kwargs)

Bases: Protocol

Grade an inbound delivery (a webhook, a posted report) from its headers and raw body.

class correspond.ops.WebhookManager(*args, **kwargs)

Bases: Protocol

Move where a channel’s platform delivers its updates.

Not an operation in OPERATIONS, but a write all the same: whoever the webhook points at reads every update the account receives, so it runs the before_send check (operation="set_webhook", the URL as the draft’s text).

class correspond.ops.WebhookReader(*args, **kwargs)

Bases: Protocol

Where a channel’s platform delivers its updates now: polling, or a webhook it POSTs to.

class correspond.ops.Writer(*args, **kwargs)

Bases: Protocol

Send a draft to a conversation.

correspond.ops.audience(ref, draft=None, *, registry=None)

Who can read a conversation, asked of its channel now. Never raises: unknown resolves to public.

A malformed reference, an unknown or planned channel, an adapter without an audience reader, and any error while computing all give scope="public", complete=False, defaulted=True, with the reason in evidence. Nothing is cached: call it again right before sending.

Return type:

Audience

correspond.ops.capabilities(channel, *, registry=None)

What a channel can do, graded, with its limits.

Return type:

Capabilities

correspond.ops.close(ref, *, reason=None, dry_run=False, registry=None, before_send=None)

Close a conversation (an issue or a discussion), with an optional reason in the channel’s vocabulary; the before_send check sees the reason (or close) as the draft’s text.

Return type:

SendResult

correspond.ops.delete_webhook(channel, *, drop_pending_updates=False, dry_run=False, registry=None, before_send=None)

Stop webhook delivery for channel, so it can be polled again; the before_send check runs (operation="delete_webhook").

Return type:

SendResult

correspond.ops.edit(ref, message_id, text, *, dry_run=False, registry=None, before_send=None)

Replace the text of a message correspond’s account wrote; the before_send check runs first, as for send().

Return type:

SendResult

correspond.ops.get_channel(name, *, registry=None)

The adapter registered under name; UnknownChannel says what to do if there is none.

Return type:

Any

correspond.ops.implemented(adapter)

The operations an adapter implements, in OPERATIONS order.

Return type:

tuple[str, ...]

correspond.ops.label(ref, labels, *, dry_run=False, registry=None, before_send=None)

Add labels to a conversation; the before_send check sees them, comma-joined, as the draft’s text.

Return type:

SendResult

correspond.ops.listen(ref, *, cursors=None, limit=None, commit=True, registry=None)

Events since the cursor stored for ref; each cursor is stored once the consumer moves past its event.

cursors defaults to files under the data root. Events can repeat after a crash or an early break: deduplicate on delivery_id.

Return type:

Iterator[Event]

correspond.ops.parse_ref(ref, *, registry=None)

A reference normalised by its channel’s adapter (kind and parent filled in, the id validated).

Return type:

ConversationRef

correspond.ops.react(ref, message_id, reaction, *, dry_run=False, registry=None, before_send=None)

Add a reaction to a message (the channel’s capabilities list the reactions it accepts); the before_send check sees the reaction as the draft’s text.

Return type:

SendResult

correspond.ops.read(ref, *, since=None, limit=None, registry=None)

The messages of a conversation, oldest first (limit keeps the most recent ones).

Return type:

list[Message]

correspond.ops.reopen(ref, *, dry_run=False, registry=None, before_send=None)

Reopen a closed conversation; the before_send check sees reopen as the draft’s text.

Return type:

SendResult

correspond.ops.send(ref, text, *, title=None, reply_to=None, priority=None, cc=(), bcc=(), native=None, notify=(), attachments=(), dry_run=False, registry=None, before_send=None, idempotency_key=None, sends=None)

Send text (or a Draft) to a conversation; dry_run shows the plan and sends nothing.

attachments (Draft.attachments) are files that go with the text, on channels that grade upload, each within its max_upload_bytes.

cc and bcc copy further recipients, on channels that grade cc (email). native carries channel-specific fields (Draft.native), checked against the channel’s capabilities.draft_fields: an unknown key is a validation failure, never silently dropped. notify names who must be notified (Draft.notify); the channel’s writer guarantees it (a mention added, an address copied) or the send fails with NotSupported("notify"). before_send(ref, draft, audience) runs first, on the dry run too; when None, the config’s before_send reference, else correspond.outbound.notice().

idempotency_key makes sending the same message again safe: a key that already sent answers that send’s result and posts nothing, and one whose earlier attempt may have gone out is confirmed by reading the conversation back, or refused as unconfirmed (correspond.idempotency). sends is where keys are kept: by default files under the data root (correspond.stores.send_store()). A mapping with a claim(key, record) -> bool (as that store has) is exclusive across processes; a plain one is checked, then set, which covers one process.

Return type:

SendResult

correspond.ops.set_webhook(channel, url, *, dry_run=False, registry=None, before_send=None)

Have channel’s platform POST its updates to url: listening by polling stops until delete_webhook().

Whoever url belongs to reads every update the account receives from then on, so the before_send check runs (operation="set_webhook", the URL as the draft’s text), on the dry run too.

Return type:

SendResult

correspond.ops.unlabel(ref, labels, *, dry_run=False, registry=None, before_send=None)

Remove labels from a conversation; the before_send check sees them, comma-joined, as the draft’s text.

Return type:

SendResult

correspond.ops.upload(ref, name, data, *, media_type='application/octet-stream', dry_run=False, source=None, native=None, registry=None, before_send=None, idempotency_key=None, sends=None)

Send a file to a conversation.

native carries the channel’s own options, the keys its capabilities.draft_fields list (Telegram: {"as_document": true} sends an image as a file, not a photo). idempotency_key keeps a retry from posting the file twice, as for send(); an upload is not read back, so one whose earlier attempt may have gone out is refused as unconfirmed.

The before_send check sees the file name as the draft’s text (operation="upload") and the file’s metadata in draft.native["upload"]: name, media_type, bytes, sha256 and, when the caller read it from disk, source (its path), so a check can hold a large, unknown or out-of-place file. It never sees the contents: a check that must vet them should hold or refuse.

Return type:

SendResult

correspond.ops.verify(channel, headers, body, *, registry=None)

Grade an inbound delivery on channel from its headers and raw body.

Return type:

Authenticity

correspond.ops.webhook_info(channel, *, registry=None)

Where channel’s platform delivers its updates now (an empty url: polling).

Return type:

dict

correspond.ops.window(messages, *, since=None, limit=None)

Messages sent or edited at or after since, oldest first, keeping the last limit.

Return type:

list[Message]

correspond.ops.with_final_cursor(events, final_cursor)

Yield events, then hand final_cursor to listen(), so a quiet poll still moves the cursor forward.

Return type:

Iterator[Event]