Key-Value

natsio.kv.KeyValue

KeyValue(
    ctx: JetStreamContext,
    bucket: str,
    stream: Stream,
    *,
    key_codec: KeyCodec | None = None,
    value_codec: ValueCodec | None = None,
)

A handle to one Key-Value bucket.

Obtain via JetStreamContext.key_value() / JetStreamContext.create_key_value(). Codecs (identity by default) transform keys and values on the way in and out — the seam that keeps codec packs a plug-in rather than a breaking change.

get async

get(key: str, *, revision: int | None = None) -> KvEntry

The live entry for key (or a specific revision).

Raises KeyNotFoundError — or its subclass KeyDeletedError when the latest revision is a marker.

put async

put(
    key: str,
    value: bytes | str,
    *,
    ttl: TTLInput | None = None,
) -> int

Store a value; returns the new revision.

ttl (a timedelta, whole seconds, or "never" — ADR-43) makes this single revision self-expire — requires the bucket's allow_msg_ttl (KeyValueConfig(allow_msg_ttl=True) or limit_marker_ttl), otherwise ConfigError.

create async

create(
    key: str,
    value: bytes | str,
    *,
    ttl: TTLInput | None = None,
) -> int

Store a value only if the key has no live value.

Succeeds for brand-new keys and for deleted/purged keys; raises KeyExistsError when a live value exists. ttl (whole seconds, or "never") makes the created revision self-expire — same allow_msg_ttl requirement as put(); when it expires the key is gone and create() succeeds for it again.

update async

update(key: str, value: bytes | str, *, last: int) -> int

Compare-and-set: store only if last is the key's latest revision.

Raises WrongLastSequenceError on conflict.

delete async

delete(key: str, *, last: int | None = None) -> None

Write a delete marker; history below it is preserved.

purge async

purge(
    key: str,
    *,
    ttl: TTLInput | None = None,
    last: int | None = None,
) -> None

Write a purge marker and roll up: prior revisions are removed.

ttl (a timedelta, whole seconds, or "never") lets the marker itself expire — requires per-message TTLs on the bucket (KeyValueConfig(allow_msg_ttl=True) or limit_marker_ttl). last makes the rollup a compare-and-set: it proceeds only if last is the key's latest revision, else WrongLastSequenceError.

purge_deletes async

purge_deletes(
    *, older_than: timedelta | None = None
) -> None

Remove delete/purge markers (and the history they sit atop).

For each marked key the whole subject is purged, so both the marker and any surviving history below it are erased. A marker YOUNGER than older_than is kept — its subject is purged only up to the marker (keep=1) so the tombstone itself survives for late watchers.

older_than defaults to 30 minutes (matching nats.go). Pass timedelta(0) (or a negative delta) to remove every marker regardless of age.

watch

watch(
    *keys: str,
    include_history: bool = False,
    updates_only: bool = False,
    ignore_deletes: bool = False,
    meta_only: bool = False,
    resume_from_revision: int | None = None,
) -> KvWatcher

Watch one or more keys (or wildcards) for changes.

With no keys the whole bucket is watched (">"); one key behaves exactly as a single-key watch; several keys install an ordered consumer with one filter subject per key, so the initial state and live updates cover the union of the keys. Yields KvEntry items and exactly one None marker once the current state has been fully delivered (immediately for updates_only); afterwards it streams live updates. Self-healing — backed by the ordered consumer.

resume_from_revision replays from that stream revision onward (every revision, not last-per-subject) and is mutually exclusive with include_history and updates_only.

keys async

keys() -> list[str]

Every key with a live value. An empty bucket returns [] promptly.

iter_keys async

iter_keys() -> AsyncGenerator[str]

Stream keys with live values without buffering the whole keyspace.

history async

history(key: str) -> list[KvEntry]

Every stored revision of key, oldest first (markers included).

natsio.kv.KvWatcher

KvWatcher(
    kv: KeyValue,
    ordered: OrderedConsumer,
    *,
    updates_only: bool,
    ignore_deletes: bool,
)

An active watch: async iterator of KvEntry | None.

The single None is the initial-state marker — everything before it existed when the watch started; everything after is a live update. Use as an async context manager for deterministic teardown.

__await__

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

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

natsio.kv.KeyValueConfig dataclass

KeyValueConfig(
    *,
    bucket: str,
    description: str | None = None,
    history: int = 1,
    ttl: timedelta | None = None,
    max_value_size: int = -1,
    max_bytes: int = -1,
    storage: StorageType = StorageType.FILE,
    replicas: int = 1,
    placement: Placement | None = None,
    republish: Republish | None = None,
    compression: bool = False,
    allow_msg_ttl: bool = False,
    limit_marker_ttl: timedelta | None = None,
    metadata: dict[str, str] | None = None,
)

Bucket configuration (maps onto an ADR-8 KV_<bucket> stream).

natsio.kv.KvEntry dataclass

KvEntry(
    bucket: str,
    key: str,
    value: bytes,
    revision: int,
    operation: Operation = Operation.PUT,
    created: datetime | None = None,
    delta: int = 0,
)

One revision of a key.

delta class-attribute instance-attribute

delta: int = 0

Revisions between this entry and the newest at read time (watch only).