Channels API for Python

Every public class, function and type of the Channels engine in tinystore, generated from its source. The Channels guide explains how to use them, and the Bun and Node, Go and Rust pages list the same API.

#Message

@dataclass(frozen=True, slots=True)
class Message[T]:
    value: T
    branch: list[str]

A message as a subscriber receives it: what was published, and the branch it was published under.

#Missed

@dataclass(frozen=True, slots=True)
class Missed:
    reason: Literal['behind', 'reconnect']
    count: int | None = None

Messages a subscriber may have lost.

It fell behind its branch's log, and count says how many; or its connection was lost and made again, and how many is unknown.

#Channel

class Channel[T]:
    name: str
    branch: tuple[str, ...]

A channel of messages of one type, or one of its branches: store.channel(name, Chat).

#Channel.under

def under(*owners: Key) -> Channel[T]: ...

The branch of owners under this one: whose messages these are.

#Channel.publish

async def publish(value: T) -> None: ...

Hands value to every subscriber of this branch that listens now, and of those above it listened to recursive.

It answers once the store has the message, not once anyone handled it.

#Channel.subscribe

async def subscribe(
    handler: Callable[[Message[T]], object],
    *,
    recursive: bool = False,
    on_missed: OnMissed | None = None,
    on_error: OnError | None = None,
) -> Subscription: ...

Calls handler with each message published from now on, one at a time and in order.

It answers once the store has the subscription, so that a message published after it arrives. The handler may be async; what it raises goes to on_error, and the store's log says it once a minute without one:

sub = await chat.under(room_id).subscribe(lambda m: show(m.value), on_missed=lambda missed: reload())
await sub.stop()

#Channel.each

async def each(
    *,
    recursive: bool = False,
    on_missed: OnMissed | None = None,
) -> AsyncIterator[Message[T]]: ...

The messages published from now on, as a loop: it listens from its first turn, and break ends it.

A raise in the loop's body ends it too:

async for m in chat.under(room_id).each():
    await save(m.value)

#Subscription

class Subscription:
    ...

A subscription whose handler is called with each message: stop ends it, and the store's close does.

#Subscription.stop

async def stop() -> None: ...

No call starts after it, and it answers once the running call ended; inside its own handler, at once.

#Subscription.start

async def start(handler: Callable[[Message[Any]], object]) -> None: ...

Registers the subscription, then runs its handler in the background.

#TxChannel

class TxChannel[T]:
    ...

A channel inside a transaction: its messages are handed over once the transaction commits.

#TxChannel.under

def under(*owners: Key) -> TxChannel[T]: ...

#TxChannel.publish

async def publish(value: T) -> None: ...