JetStream

natsio.jetstream.JetStreamContext

JetStreamContext(
    client: Client,
    *,
    domain: str | None = None,
    api_prefix: str | None = None,
    timeout: float = 5.0,
    publish_async_max_pending: int = _ASYNC_MAX_PENDING,
    publish_async_stall_wait: float = _ASYNC_STALL_WAIT,
    publish_async_timeout: float | None = None,
)

Entry point to JetStream. Obtain via Client.jetstream().

publish_async_pending property

publish_async_pending: int

Number of async publishes still awaiting their PubAck.

api_level async

api_level() -> int

The server's advertised JetStream API level (nats-server 2.14.3 reports 4).

create_or_update_stream async

create_or_update_stream(config: StreamConfig) -> Stream

Update the stream, creating it when absent (nats.go CreateOrUpdateStream).

The idempotent way to assert a stream in scripts and services: re-running never raises StreamNameInUseError — an existing stream is updated to config (a no-op when identical), a missing one is created. Update-then-create mirrors nats.go's order, and a create that loses a race to a concurrent creator is absorbed by a re-update, so the call stays idempotent under contention.

stream async

stream(name: str) -> Stream

A handle to an existing stream (fetches and caches its info).

purge_stream async

purge_stream(
    name: str,
    *,
    subject: str | None = None,
    sequence: int | None = None,
    keep: int | None = None,
) -> int

Purge messages; returns the number purged.

create_key_value async

create_key_value(
    config: KeyValueConfig,
    *,
    key_codec: KeyCodec | None = None,
    value_codec: ValueCodec | None = None,
) -> KeyValue

Create a Key-Value bucket.

Re-creating with an identical configuration is idempotent; an existing bucket with a DIFFERENT configuration raises BucketExistsError.

update_key_value async

update_key_value(
    config: KeyValueConfig,
    *,
    key_codec: KeyCodec | None = None,
    value_codec: ValueCodec | None = None,
) -> KeyValue

Update an existing Key-Value bucket's configuration.

Runs a STREAM.UPDATE on the mapped backing stream. Raises BucketNotFoundError when no such bucket exists — there is nothing to update.

create_or_update_key_value async

create_or_update_key_value(
    config: KeyValueConfig,
    *,
    key_codec: KeyCodec | None = None,
    value_codec: ValueCodec | None = None,
) -> KeyValue

Create the bucket, or update it in place if it already exists.

Mirrors nats.go CreateOrUpdateStream: attempt the update first and fall back to a create when the backing stream is absent. Race-tolerant under concurrent creators (see create_or_update_stream()).

key_value async

key_value(
    bucket: str,
    *,
    key_codec: KeyCodec | None = None,
    value_codec: ValueCodec | None = None,
) -> KeyValue

A handle to an existing Key-Value bucket.

key_value_store_names async

key_value_store_names() -> AsyncIterator[str]

Yield the name of every Key-Value bucket (KV_ prefix stripped).

Pages STREAM.NAMES filtered to the $KV.*.> keyspace, then keeps only streams carrying the KV_ name prefix — excluding plain streams that merely publish on $KV subjects (matching nats.go's dual filter).

key_value_stores async

key_value_stores() -> AsyncIterator[KeyValueStatus]

Yield a KeyValueStatus for every KV bucket.

Statuses are built straight from the paged STREAM.LIST info, so no extra per-bucket round-trip is made. Filtering matches key_value_store_names().

create_object_store async

create_object_store(
    config: ObjectStoreConfig,
) -> ObjectStore

Create an Object Store bucket.

Re-creating with an identical configuration is idempotent; an existing bucket with a DIFFERENT configuration raises BucketExistsError.

update_object_store async

update_object_store(
    config: ObjectStoreConfig,
) -> ObjectStore

Update an existing Object Store bucket's configuration.

Runs a STREAM.UPDATE on the mapped backing stream. Raises BucketNotFoundError when no such bucket exists. The server rejects updates that would violate a sealed stream or otherwise change immutable config; those surface as their mapped APIError.

create_or_update_object_store async

create_or_update_object_store(
    config: ObjectStoreConfig,
) -> ObjectStore

Create the bucket, or update it in place if it already exists.

Mirrors nats.go CreateOrUpdateStream: update first, create on absence. Race-tolerant under concurrent creators (see create_or_update_stream()).

object_store async

object_store(bucket: str) -> ObjectStore

A handle to an existing Object Store bucket.

object_store_names async

object_store_names() -> AsyncIterator[str]

Yield the name of every Object Store bucket (OBJ_ prefix stripped).

Pages STREAM.NAMES filtered to the $O.*.C.> chunk keyspace, then keeps only streams carrying the OBJ_ name prefix — excluding plain streams that merely publish on $O subjects (matching nats.go's dual filter).

object_stores async

object_stores() -> AsyncIterator[ObjectStoreStatus]

Yield an ObjectStoreStatus for every bucket.

Statuses are built straight from the paged STREAM.LIST info, so no extra per-bucket round-trip is made. Filtering matches object_store_names().

publish async

publish(
    subject: str,
    payload: bytes | str = b"",
    *,
    headers: HeadersInput | None = None,
    msg_id: str | None = None,
    expected_stream: str | None = None,
    expected_last_seq: int | None = None,
    expected_last_subject_seq: int | None = None,
    expected_last_subject_seq_subject: str | None = None,
    expected_last_msg_id: str | None = None,
    ttl: TTLInput | None = None,
    timeout: float | None = None,
) -> PubAck

Publish to a stream and await its PubAck.

ttl is a timedelta, whole seconds, or "never" (ADR-43) and needs the stream's allow_msg_ttl. A 503 (no stream bound / leader election in progress) is retried briefly per ADR-22 before raising NoStreamResponseError.

expected_last_subject_seq_subject (server 2.12+) scopes an expected_last_subject_seq check to a different subject filter — e.g. publish to a.1 while asserting the last sequence on a.*.

publish_async async

publish_async(
    subject: str,
    payload: bytes | str = b"",
    *,
    headers: HeadersInput | None = None,
    msg_id: str | None = None,
    expected_stream: str | None = None,
    expected_last_seq: int | None = None,
    expected_last_subject_seq: int | None = None,
    expected_last_subject_seq_subject: str | None = None,
    expected_last_msg_id: str | None = None,
    ttl: TTLInput | None = None,
) -> asyncio.Future[PubAck]

Publish without waiting; return a future that resolves to the PubAck.

The message is sent immediately with a per-message reply inbox; the server's ack is routed back to the returned future. Await the future to get the PubAck or the failure — an APIError (e.g. WrongLastSequenceError), NoStreamResponseError after 503 retries are exhausted, AsyncPublishTimeoutError, or ConnectionClosedError if the connection drops with the ack still outstanding.

Outstanding acks are capped at publish_async_max_pending. When the window is full this call waits up to publish_async_stall_wait for it to drain, then raises TooManyStalledMsgsError.

Header/expectation arguments behave exactly as publish().

publish_async_complete async

publish_async_complete(
    timeout: float | None = None,
) -> None

Wait until every outstanding async publish has resolved (ack or error).

Returns immediately when the window is already empty. With timeout, raises TimeoutError if the window has not drained in time; the outstanding publishes keep their own futures.

natsio.jetstream.Stream

Stream(ctx: JetStreamContext, info: StreamInfo)

A thin, stateful handle to one stream.

cached_info is populated when the handle is created and refreshed only by an explicit info() call.

create_consumer async

create_consumer(
    config: ConsumerConfig | None = None,
) -> Consumer

Create (or idempotently assert) a consumer and return its handle.

A durable_name (or name) makes it durable; otherwise a name is generated client-side and the consumer is ephemeral (set inactive_threshold to bound its lifetime).

consumer async

consumer(name: str) -> Consumer

A handle to an existing consumer (fetches and caches its info).

pause_consumer async

pause_consumer(name: str, until: datetime) -> None

Pause delivery until until (server 2.11+).

ordered_consumer

ordered_consumer(
    *,
    filter_subjects: list[str] | None = None,
    deliver_policy: DeliverPolicy = DeliverPolicy.ALL,
    opt_start_seq: int | None = None,
    opt_start_time: datetime | None = None,
    headers_only: bool = False,
) -> OrderedConsumer

A single-threaded, always-in-order view of the stream (ADR-17).

Ephemeral and self-healing: on a gap, a missed heartbeat, or consumer loss it recreates itself at the next unseen stream sequence.

get_msg async

get_msg(
    sequence: int | None = None,
    *,
    subject: str | None = None,
    next_for: bool = False,
) -> StoredMsg

Read one stored message by sequence or (last-by / next-from) subject.

Uses Direct Get when the stream allows it (the 2.14-era default read path), else the STREAM.MSG.GET API.

get_last_msgs_for async

get_last_msgs_for(
    subjects: list[str] | str,
    *,
    batch: int | None = None,
    up_to_seq: int | None = None,
    up_to_time: datetime | None = None,
) -> AsyncIterator[StoredMsg]

Yield the last stored message on each of subjects in one batch.

A single batch Direct Get (multi_last, ADR-31): one request to $JS.API.DIRECT.GET.<stream> streams back the last message per matched subject, terminated by a 204 EOB frame. subjects may contain wildcards (* / >); a subject with no stored message is simply omitted. This is nats.go / orbit GetLastMsgsFor parity.

The batch API is Direct-Get-only: the stream must have been created with allow_direct=True, else ConfigError is raised (there is no STREAM.MSG.GET fallback for the batch form).

• batch caps how many messages the server returns.
• up_to_seq / up_to_time fetch the last message on each subject at or before that stream sequence / timestamp (mutually exclusive).

Ordinary usage returns nothing to await up front — iterate it:

async for stored in stream.get_last_msgs_for(["a.*", "b.>"]):
    ...

natsio.jetstream.Consumer

Consumer(stream: Stream, info: ConsumerInfo)

A handle to a pull consumer. Spawns fetches and consume sessions.

unpin async

unpin(group: str) -> None

Release the pinned client for group (ADR-42 pinned_client).

The server drops the current pin; the next pull request from any client in the group becomes the new pinned client. Raises ConsumerNotFoundError if the consumer no longer exists.

fetch async

fetch(
    max_messages: int = 100,
    *,
    max_bytes: int | None = None,
    timeout: float = 5.0,
    idle_heartbeat: float | None = None,
    no_wait: bool = False,
    group: str | None = None,
    min_pending: int | None = None,
    min_ack_pending: int | None = None,
) -> list[JsMsg]

One bounded pull: up to max_messages within timeout seconds.

Returns what arrived when the batch fills, the server ends the request (no more messages / request expired), or the deadline passes — an empty list is a normal outcome. no_wait returns only immediately-available messages. Raises ConnectionClosedError if the connection closes mid-fetch with the batch still incomplete.

group selects an ADR-42 priority group (required, and validated, when the consumer is configured with priority groups). min_pending / min_ack_pending gate delivery on an overflow-policy consumer: the server only serves this request when the consumer has at least that many pending messages / unacked messages (server-enforced).

next async

next(
    *,
    timeout: float = 5.0,
    group: str | None = None,
    min_pending: int | None = None,
    min_ack_pending: int | None = None,
) -> JsMsg

The next available message, or NoMessagesError on expiry.

group / min_pending / min_ack_pending carry the same ADR-42 priority-group semantics as fetch().

consume

consume(
    *,
    max_messages: int = 500,
    max_bytes: int | None = None,
    expires: float = 30.0,
    idle_heartbeat: float | None = None,
    threshold: float = 0.5,
    group: str | None = None,
    min_pending: int | None = None,
    min_ack_pending: int | None = None,
) -> Consumption

A continuous, self-refilling message stream (ADR-37 consume).

Keeps up to max_messages requested from the server, re-pulling when the outstanding count drops below threshold of the target so the buffer never drains to a stall. Use as an async context manager:

async with consumer.consume() as messages:
    async for msg in messages:
        await msg.ack()

group / min_pending / min_ack_pending carry ADR-42 priority- group semantics (see fetch()); the group, when the consumer has priority groups configured, is validated here before the session starts.

natsio.jetstream.OrderedConsumer

OrderedConsumer(stream: Stream, base: ConsumerConfig)

An always-in-order, self-healing stream view (ADR-17).

Backed by an ephemeral pull consumer with ack_policy=none. Order is judged by consumer sequence contiguity; on any gap, stall, or consumer loss it silently recreates itself starting at the next unseen stream sequence (deliver_policy=by_start_sequence — always paired with opt_start_seq).

Prefer the context-manager form for deterministic teardown of the server-side ephemeral consumer:

async with stream.ordered_consumer() as ordered:
    async for msg in ordered:
        ...

A bare async for over the object also works; teardown then happens at generator finalization (or via stop()).

__await__

__await__() -> Generator[None, None, OrderedConsumer]

await is optional and completes immediately (no I/O) — nats-py muscle memory support; see Subscription.__await__().

start async

start() -> ConsumerInfo

Eagerly create the underlying ephemeral consumer.

Iteration does this lazily; starting explicitly is useful when the caller needs the initial ConsumerInfo (e.g. num_pending to detect an initially-empty stream) before blocking on messages.

stop async

stop() -> None

Deterministically stop the session and delete the ephemeral consumer.

messages

messages(
    *,
    max_messages: int = 500,
    expires: float = 30.0,
    idle_heartbeat: float | None = None,
    idle_timeout: float | None = None,
    until_drained: bool = False,
) -> AsyncGenerator[JsMsg]

The ordered message stream (an async generator — aclose() to stop).

idle_timeout bounds the wait for each message: on expiry the stream raises NoMessagesError instead of silently self-healing forever, letting callers distinguish a quiet stream from a dead one.

until_drained=True is the finite-read mode: iteration ends normally once the consumer is caught up (the server's per-delivery num_pending reaches zero — exact and immediate, no timeout wait). Messages published while draining are still delivered; "caught up" means caught up at delivery time. An empty stream yields nothing. Combine with idle_timeout if you also want a liveness bound while draining (it raises, as above).

natsio.jetstream.JsMsg

JsMsg(msg: Msg, client: Client)

A message delivered by a JetStream consumer.

Wraps the core Msg and adds acknowledgement. ack/nak/term are terminal — sending a second terminal ack raises MessageAlreadyAckedError; in_progress may be sent any number of times before a terminal ack.

ack async

ack() -> None

Acknowledge successful processing.

ack_sync async

ack_sync(timeout: float | None = 5.0) -> None

Acknowledge and wait for the server to confirm it was recorded.

nak async

nak(*, delay: timedelta | float | None = None) -> None

Negative-acknowledge: redeliver (optionally after delay).

term async

term(reason: str = '') -> None

Terminate delivery: never redeliver this message.

in_progress async

in_progress() -> None

Reset the ack-wait timer; the message is still being worked on.

natsio.jetstream.StreamConfig dataclass

StreamConfig(
    *,
    extra: dict[str, Any] = dict(),
    name: str = "",
    description: str | None = None,
    subjects: list[str] | None = None,
    retention: RetentionPolicy = RetentionPolicy.LIMITS,
    max_consumers: int = -1,
    max_msgs: int = -1,
    max_bytes: int = -1,
    max_age: Annotated[
        timedelta | None, NS_DURATION
    ] = None,
    max_msgs_per_subject: int = -1,
    max_msg_size: int = -1,
    discard: DiscardPolicy = DiscardPolicy.OLD,
    discard_new_per_subject: bool | None = None,
    storage: StorageType = StorageType.FILE,
    num_replicas: int = 1,
    no_ack: bool | None = None,
    duplicate_window: Annotated[
        timedelta | None, NS_DURATION
    ] = None,
    placement: Placement | None = None,
    mirror: StreamSource | None = None,
    sources: list[StreamSource] | None = None,
    sealed: bool | None = None,
    deny_delete: bool | None = None,
    deny_purge: bool | None = None,
    allow_rollup_hdrs: bool | None = None,
    compression: StorageCompression | None = None,
    first_seq: int | None = None,
    subject_transform: SubjectTransform | None = None,
    republish: Republish | None = None,
    allow_direct: bool | None = None,
    mirror_direct: bool | None = None,
    consumer_limits: ConsumerLimits | None = None,
    metadata: dict[str, str] | None = None,
    allow_msg_ttl: bool | None = None,
    subject_delete_marker_ttl: Annotated[
        timedelta | None, NS_DURATION
    ] = None,
    allow_atomic: bool | None = None,
    allow_msg_counter: bool | None = None,
    allow_msg_schedules: bool | None = None,
)

Bases: JsonModel

natsio.jetstream.ConsumerConfig dataclass

ConsumerConfig(
    *,
    extra: dict[str, Any] = dict(),
    name: str | None = None,
    durable_name: str | None = None,
    description: str | None = None,
    deliver_policy: DeliverPolicy = DeliverPolicy.ALL,
    opt_start_seq: int | None = None,
    opt_start_time: Annotated[
        datetime | None, RFC3339
    ] = None,
    ack_policy: AckPolicy = AckPolicy.EXPLICIT,
    ack_wait: Annotated[
        timedelta | None, NS_DURATION
    ] = None,
    max_deliver: int | None = None,
    backoff: list[Annotated[timedelta, NS_DURATION]]
    | None = None,
    filter_subject: str | None = None,
    filter_subjects: list[str] | None = None,
    replay_policy: ReplayPolicy = ReplayPolicy.INSTANT,
    sample_freq: str | None = None,
    max_waiting: int | None = None,
    max_ack_pending: int | None = None,
    headers_only: bool | None = None,
    max_batch: int | None = None,
    max_expires: Annotated[
        timedelta | None, NS_DURATION
    ] = None,
    max_bytes: int | None = None,
    inactive_threshold: Annotated[
        timedelta | None, NS_DURATION
    ] = None,
    num_replicas: int | None = None,
    mem_storage: bool | None = None,
    metadata: dict[str, str] | None = None,
    priority_groups: list[str] | None = None,
    priority_policy: PriorityPolicy | None = None,
    priority_timeout: Annotated[
        timedelta | None, NS_DURATION
    ] = None,
    pause_until: Annotated[datetime | None, RFC3339] = None,
)

Bases: JsonModel

Pull-consumer configuration (ADR-37: pull is the only delivery model).

Push-only fields a foreign consumer might carry survive round-trips via extra without being modeled here.

natsio.jetstream.PubAck dataclass

PubAck(
    *,
    extra: dict[str, Any] = dict(),
    stream: str = "",
    seq: int = 0,
    domain: str | None = None,
    duplicate: bool | None = None,
    val: str | None = None,
    batch_id: str | None = None,
    batch_size: int | None = None,
)

Bases: JsonModel