temporalio.contrib.workflow_streams.TopicHandle

class documentation

class TopicHandle(Generic[T]): (source)

Constructor: TopicHandle(client, name, type)

Client-side handle for publishing to and subscribing from a single topic.

Warning

This class is experimental and may change in future versions.

Constructed via WorkflowStreamClient.topic. Publishes share the underlying client's batching, dedup, and codec path; this object holds only the topic name and bound type.

Method	`__init__`	Bind the handle to a client, topic name, and type.
Method	`publish`	Buffer `value` for publishing on this topic.
Async Method	`subscribe`	Async iterator over items on this topic, decoded as `T`.
Property	`name`	The topic name this handle is bound to.
Property	`type`	The value type this handle is bound to.
Instance Variable	`_client`	Undocumented
Instance Variable	`_name`	Undocumented
Instance Variable	`_type`	Undocumented

def __init__(self, client: WorkflowStreamClient, name: str, type: type[T]): (source) ¶

Bind the handle to a client, topic name, and type.

Prefer WorkflowStreamClient.topic over calling this directly; the factory is what records the per-client type binding and rejects conflicts.

def publish(self, value: T | Payload, *, force_flush: bool = False): (source) ¶

Buffer value for publishing on this topic.

Equivalent to the underlying client's publish path; the value flows through the same buffer, batch interval, and dedup sequence.

Parameters
value:`T \| Payload`	Value to publish. Goes through the client's sync payload converter at flush time. A pre-built `temporalio.api.common.v1.Payload` bypasses conversion (zero-copy fast path), regardless of the handle's bound type.
force_flush:`bool`	If True, wake the flusher to send immediately (fire-and-forget — does not block the caller).

async def subscribe(self, from_offset: int = 0, *, poll_cooldown: timedelta = timedelta(milliseconds=100)) -> AsyncIterator[WorkflowStreamItem[T]]: (source) ¶

Async iterator over items on this topic, decoded as T.

For raw Payload access, or any other decode type that differs from the handle's bound T, use WorkflowStreamClient.subscribe directly with an explicit result_type (typically temporalio.common.RawValue). The handle's bound type intentionally cannot be Payload — the converter has no Payload decode path.

Parameters
from_offset:`int`	Global offset to start reading from.
poll_cooldown:`timedelta`	Minimum interval between polls when there are no new items.

Returns
`AsyncIterator[WorkflowStreamItem[T]]`	Undocumented

@property

name: str = (source) ¶

The topic name this handle is bound to.

@property

type: type[T] = (source) ¶

The value type this handle is bound to.

_client = (source) ¶

Undocumented

_name = (source) ¶

Undocumented

_type = (source) ¶

Undocumented