Live streams¶

Short-lived streams of payloads by key, read by every current reader: a live view of something that happens now, e.g. the progress of a job or the tokens of an answer. Delivery is at most once and not durable: the state a stream shows is stored elsewhere.

Quickstart¶

from pydantic import BaseModel

from commons.streams import InMemoryLiveStreams, PydanticCodec, TypedLiveStreams


class Progress(BaseModel):
    step: str
    done: int


live_streams = InMemoryLiveStreams()  # one backend per application
progress = TypedLiveStreams(live_streams, "myapp:progress", PydanticCodec[Progress](Progress))

# The worker:
await progress.append(job_id, [Progress(step="load", done=1)])
await progress.close(job_id)

# Any number of readers, e.g. an SSE endpoint:
async for position, value in progress.read(job_id): ...

Concepts¶

Live streams (LiveStreams): streams of byte payloads, by key. Writers append; every reader gets every payload from where it starts reading. One backend serves all kinds of streams.
Typed live streams (TypedLiveStreams[T] with a Codec[T]): a type-safe view on one kind of stream: values of T by id, under a key prefix, encoded e.g. as JSON with PydanticCodec.
Position: an opaque string per payload, increasing within a stream, e.g. the id of an SSE event, so a reader resumes with Last-Event-ID.

Configuration¶

InMemoryLiveStreams(retention=300.0): seconds a closed stream stays readable, for late readers and reconnects.

Usage¶

Choose a prefix of your own (e.g. myapp:progress), so streams of different kinds, also those of libraries, do not collide on one backend.
Close the stream when the work is done; readers then stop after its last payload.
Resume a reader with the last position it saw: read(id, after=position).
Encode several kinds of values in one stream with a discriminated union: PydanticCodec[Event](Annotated[Started | Progressed, Field(discriminator="kind")]); by_alias=False encodes field names instead of aliases.
Don't wait for a stream that may never come: read(id, wait=False) ends at once if the stream doesn't exist.
Treat streaming as best effort. If appending fails, the work goes on and its result is stored; only the live view is missing.

Behavior¶

From the start, or resumed. Without after, a reader gets all payloads of the stream, also when it connects late; with after, only those after that position.
Missing streams are waited for, unless wait=False: a reader may connect before the writer started.
The stream ends for its readers when it is closed. A closed stream is kept for retention seconds, then disappears; reading it then waits for a new stream, or ends with wait=False.
Single application instance. InMemoryLiveStreams only serves readers in the same process; a Redis adapter would serve several instances.

Errors¶

Exception	Raised when	What to do
`ValueError`	`append` to a closed stream.	Use a new key for new work.
`ValueError` (while reading)	`after` is not a position of this kind of stream.	Answer 400, or read without `after`.

API reference¶

commons.streams ¶

Live streams: short-lived streams of payloads by key, fanned out to every current reader, at most once.

Port (LiveStreams) and its in-process adapter (InMemoryLiveStreams, single application instance).
Typed view (TypedLiveStreams with a Codec, e.g. PydanticCodec): values of one type under a key prefix.

Not durable: a live stream shows state that is stored elsewhere, e.g. the progress of a job, while it happens.

Modules:

Name	Description
`live_streams`
`typed`

Classes:

Name	Description
`InMemoryLiveStreams`	A `LiveStreams` in process memory: readers on the same application instance only.
`LiveStreams`	Short-lived streams of opaque payloads, by key, read by every current reader (fan-out).
`Codec`	Turns values into the payloads of a `LiveStreams` stream and back.
`PydanticCodec`	Encodes values as JSON with a Pydantic `TypeAdapter`: models, and discriminated unions of models.
`TypedLiveStreams`	A type-safe view on one kind of stream of a `LiveStreams`: values of `T`, by id, under a key prefix.

all `module-attribute` ¶

__all__ = [
    "Codec",
    "InMemoryLiveStreams",
    "LiveStreams",
    "PydanticCodec",
    "TypedLiveStreams",
]

InMemoryLiveStreams ¶

InMemoryLiveStreams(*, retention: float = 300.0)

A LiveStreams in process memory: readers on the same application instance only.

Payloads are kept as the bytes they were appended as, so values go through their codec as with any other backend. The position of a payload is its number in the stream, starting at "1". A closed stream is kept for retention seconds, then dropped on the next access. A Redis adapter would replace this for several instances.

Parameters:

Name	Type	Description	Default
`retention`	`float`	Seconds a closed stream stays readable.	`300.0`

Methods:

Name	Description
`append`
`close`
`read`

Source code in src/commons/streams/live_streams.py

def __init__(self, *, retention: float = 300.0) -> None:
    self._retention = retention
    self._streams: dict[str, _Stream] = {}
    self._waiters: set[asyncio.Event] = set()

append `async` ¶

append(key: str, payloads: list[bytes]) -> None

Source code in src/commons/streams/live_streams.py

async def append(self, key: str, payloads: list[bytes]) -> None:
    self._prune()
    stream = self._streams.setdefault(key, _Stream())
    if stream.closed_at is not None:
        raise ValueError(f"The stream `{key}` is closed.")

    stream.payloads.extend(payloads)
    self._notify()

close `async` ¶

close(key: str) -> None

Source code in src/commons/streams/live_streams.py

async def close(self, key: str) -> None:
    stream = self._streams.setdefault(key, _Stream())
    if stream.closed_at is None:
        stream.closed_at = time.monotonic()
        self._notify()

read `async` ¶

read(
    key: str, after: str | None = None, *, wait: bool = True
) -> AsyncIterator[tuple[str, bytes]]

Source code in src/commons/streams/live_streams.py

async def read(self, key: str, after: str | None = None, *, wait: bool = True) -> AsyncIterator[tuple[str, bytes]]:
    position = _position(after)

    while True:
        self._prune()
        stream = self._streams.get(key)
        if stream is None and not wait:
            return

        if stream is not None:
            while position < len(stream.payloads):
                position += 1
                yield str(position), stream.payloads[position - 1]

            if stream.closed_at is not None:
                return

        await self._changed()

LiveStreams ¶

Bases: Protocol

Short-lived streams of opaque payloads, by key, read by every current reader (fan-out).

Writers append payloads to a stream; any number of readers, on any application instance, read them from a position on. Streams are not durable: they carry live views, such as the events of a running job, while the state they show is stored elsewhere. A closed stream stays readable for a while, for late readers and reconnects, and then disappears. Use TypedLiveStreams for a type-safe view on one kind of stream.

Positions are opaque strings, increasing within a stream, e.g. to resume a reader with SSE's Last-Event-ID.

Methods:

Name	Description
`append`	Append the payloads to the stream, which is created by the first append.
`close`	Mark the stream finished: readers stop after its last payload. Closing twice is a no-op.
`read`	Return the payloads of the stream after the position `after` (all if None), then the new ones live, each

append `async` ¶

append(key: str, payloads: list[bytes]) -> None

Append the payloads to the stream, which is created by the first append.

Raises:

Type	Description
`ValueError`	If the stream is closed.

Source code in src/commons/streams/live_streams.py

async def append(self, key: str, payloads: list[bytes]) -> None:
    """Append the payloads to the stream, which is created by the first append.

    Raises:
        ValueError: If the stream is closed.
    """
    ...

close `async` ¶

close(key: str) -> None

Mark the stream finished: readers stop after its last payload. Closing twice is a no-op.

Source code in src/commons/streams/live_streams.py

async def close(self, key: str) -> None:
    """Mark the stream finished: readers stop after its last payload. Closing twice is a no-op."""
    ...

read ¶

read(
    key: str, after: str | None = None, *, wait: bool = True
) -> AsyncIterator[tuple[str, bytes]]

Return the payloads of the stream after the position after (all if None), then the new ones live, each with its position, until the stream is closed.

With wait, a stream that does not exist yet is waited for; without, the iterator ends at once.

Raises:

Type	Description
`ValueError`	If `after` is not a position of this kind of stream.

Source code in src/commons/streams/live_streams.py

def read(self, key: str, after: str | None = None, *, wait: bool = True) -> AsyncIterator[tuple[str, bytes]]:
    """Return the payloads of the stream after the position ``after`` (all if None), then the new ones live, each
    with its position, until the stream is closed.

    With ``wait``, a stream that does not exist yet is waited for; without, the iterator ends at once.

    Raises:
        ValueError: If ``after`` is not a position of this kind of stream.
    """
    ...

Codec ¶

Bases: Protocol

Turns values into the payloads of a LiveStreams stream and back.

Methods:

Name	Description
`encode`
`decode`

encode ¶

encode(value: T) -> bytes

Source code in src/commons/streams/typed.py

def encode(self, value: T) -> bytes: ...

decode ¶

decode(data: bytes) -> T

Source code in src/commons/streams/typed.py

def decode(self, data: bytes) -> T: ...

PydanticCodec ¶

PydanticCodec(type_: Any, *, by_alias: bool = True)

Encodes values as JSON with a Pydantic TypeAdapter: models, and discriminated unions of models.

Parameters:

Name	Type	Description	Default
`type_`	`Any`	The type to validate decoded values with, e.g. a model or a discriminated union of models; it may be wider than `T` (decoding a union member as its base class).	required
`by_alias`	`bool`	Whether to encode fields by their alias, e.g. camelCase names.	`True`

Example

class Started(BaseModel):
    kind: Literal["started"] = "started"


class Progressed(BaseModel):
    kind: Literal["progressed"] = "progressed"
    percent: int


Progress = Annotated[Started | Progressed, Field(discriminator="kind")]

codec = PydanticCodec[Started | Progressed](Progress)
codec.encode(Progressed(percent=50))  # b'{"kind":"progressed","percent":50}'

Methods:

Name	Description
`encode`
`decode`

Source code in src/commons/streams/typed.py

def __init__(self, type_: Any, *, by_alias: bool = True) -> None:
    self._adapter: TypeAdapter[T] = TypeAdapter(type_)
    self._by_alias = by_alias

encode ¶

encode(value: T) -> bytes

Source code in src/commons/streams/typed.py

def encode(self, value: T) -> bytes:
    return self._adapter.dump_json(value, by_alias=self._by_alias)

decode ¶

decode(data: bytes) -> T

Source code in src/commons/streams/typed.py

def decode(self, data: bytes) -> T:
    return self._adapter.validate_json(data)

TypedLiveStreams ¶

TypedLiveStreams(
    streams: LiveStreams, prefix: str, codec: Codec[T]
)

A type-safe view on one kind of stream of a LiveStreams: values of T, by id, under a key prefix.

The stream of id is the LiveStreams stream f"{prefix}:{id}", so several kinds of streams, also an application's own, can share one backend.

Parameters:

Name	Type	Description	Default
`streams`	`LiveStreams`	The backend.	required
`prefix`	`str`	The key prefix of this kind of stream, e.g. `"myapp:progress"`.	required
`codec`	`Codec[T]`	Encodes and decodes the values.	required

Methods:

Name	Description
`append`	Append the values to the stream of `id`; see `LiveStreams.append`.
`close`	Close the stream of `id`; see `LiveStreams.close`.
`read`	Read the stream of `id`, decoded; see `LiveStreams.read`.

Source code in src/commons/streams/typed.py

def __init__(self, streams: LiveStreams, prefix: str, codec: Codec[T]) -> None:
    self._streams = streams
    self._prefix = prefix
    self._codec = codec

append `async` ¶

append(id: str, values: list[T]) -> None

Append the values to the stream of id; see LiveStreams.append.

Source code in src/commons/streams/typed.py

async def append(self, id: str, values: list[T]) -> None:
    """Append the values to the stream of ``id``; see `LiveStreams.append`."""

    await self._streams.append(self._key(id), [self._codec.encode(value) for value in values])

close `async` ¶

close(id: str) -> None

Close the stream of id; see LiveStreams.close.

Source code in src/commons/streams/typed.py

async def close(self, id: str) -> None:
    """Close the stream of ``id``; see `LiveStreams.close`."""

    await self._streams.close(self._key(id))

read `async` ¶

read(
    id: str, after: str | None = None, *, wait: bool = True
) -> AsyncIterator[tuple[str, T]]

Read the stream of id, decoded; see LiveStreams.read.

Source code in src/commons/streams/typed.py

async def read(self, id: str, after: str | None = None, *, wait: bool = True) -> AsyncIterator[tuple[str, T]]:
    """Read the stream of ``id``, decoded; see `LiveStreams.read`."""

    async for position, payload in self._streams.read(self._key(id), after, wait=wait):
        yield position, self._codec.decode(payload)

Live streams¶

Quickstart¶

Concepts¶

Configuration¶

Usage¶

Behavior¶

Errors¶

API reference¶

commons.streams ¶

__all__ module-attribute ¶

InMemoryLiveStreams ¶

append async ¶

close async ¶

read async ¶

LiveStreams ¶

append async ¶

close async ¶

read ¶

Codec ¶

encode ¶

decode ¶

PydanticCodec ¶

encode ¶

decode ¶

TypedLiveStreams ¶

append async ¶

close async ¶

read async ¶

all `module-attribute` ¶

append `async` ¶

close `async` ¶

read `async` ¶

append `async` ¶

close `async` ¶

append `async` ¶

close `async` ¶

read `async` ¶