3tears-nats
Typed NATS client wrapper, subject builders, and JetStream KV bucket primitives for 3tears applications.
What this package provides
NatsClient-- single canonical wrapper aroundnats-py. Handles connect (with a bounded startup timeout, then unbounded runtime reconnect -- it rides out an outage of any length rather than closing itself), graceful shutdown/drain, typed publish, kw-only subscribe with optional Pydantic validation, request/reply withtimedeltatimeouts, JetStream KV bucket access, and stream declaration (ensure_jetstream_stream). Streams and KV buckets the client declared (ensure_jetstream_stream,ensure_kv_bucket), on memory or file storage, and the durable consumers it bound on them, are created again after every reconnect, so a NATS restart that wipes them -- memory storage always, file storage when the restart lost its volume, as it can on Kubernetes -- heals without a process restart. They come back empty; contents are their owner's to write again. To have one refilled, declare it withensure_kv_bucket(on_restored=...), which the client runs with the live handle each time it creates the bucket again, or letPersistedCopyBucketown it.Subject+Subjects-- opaque subject dataclass and factory of every canonical subject family used by 3tears applications. Replaces ad-hocf"{namespace}.tools.call"string-concatenation across the platform.NatsKvBucket-- operations against one JetStream KV bucket (get/put/delete/create/update/get_entry/list_keys). Bucket name auto-prefixed by the connected client'snats_subject_namespace, unless it was declared withensure_kv_bucket(prefix_namespace=False), which uses the name exactly. Every operation follows the client onto its current connection, so a handle survives a credential renewal or a move off a lame-duck server.PersistedCopyBucket-- the one owner of a KV bucket that is the persisted copy of an in-memory structure. It declares the bucket under its exact name in the background, retrying until the declaration lands; runsload(bucket into memory) once, at the first declaration; and runswrite_back(memory into bucket) then and every time the client creates the bucket again empty. Use it rather than hand-rolling a declare/load/refill loop.NatsObjectStore/ObjectInfo-- one JetStream Object Store bucket: immutable named objects, chunked on the wire in the NATS Object Store's own shape. See Object Store.CopyWriteHealth-- whether writes to a persisted copy are landing, counted per write operation, as a readiness probe reads it (persisting).forward/serve_owner-- payload-agnostic owner-routed request/reply: send a request to whichever pod currently serves a key and get its reply back. A separate election mechanism decides who owns the key; this only carries the message.attach_pipe/serve_pipe/open_pipe-- a payload-agnostic byte pipe to whichever pod owns a key, for reaching a process that has no inbound network path. Rendezvous ridesforward; the stream then moves to its own subjects with a sequenced framing (a lost frame raises rather than being skipped) and a credit window that stops the producer reading its source when the consumer falls behind.StreamTransport-- narrow Protocol used by streaming consumers; lets test fakes substitute for the live client.- Errors --
NatsClientError,SubscribeError,PublishError,RequestError,KvError, andKvBucketNotFoundError(aKvError). See KV errors. The Object Store raisesObjectStoreError, withObjectStoreNotFoundError,ObjectNotFoundError,ObjectExistsErrorandObjectStoreFullErrorfor the cases a caller answers differently. is_bucket_not_found/is_key_not_found/is_nats_error-- classify the failures of a RAW nats-py handle by type, so a consumer that keepsnats.*imports out of its code never matches nats-py class names as strings.
Why a separate package
The wrapper is consumed by the platform services (broker, gateway, registry, channel adapters, agent SDK) and any 3tears-based application. Keeping it in 3tears-nats avoids forcing those apps to depend on a host application repo just for a NATS primitive.
Mistake-proofed API
- Subscribe is keyword-only after
self. A common production bug (nc.subscribe(subject, callback)silently treating the callback as a queue group innats-py2.10+) is impossible to reproduce against this wrapper. - Publish accepts
BaseModelinstances. Raw bytes go through the explicit escape hatchpublish_raw. - Subjects are typed
Subjectobjects, not strings. The factory owns subject formatting; callers cannot accidentally interpolate the wrong shape. - Default
deadletter_on_failure=True. Uncaught subscribe-callback exceptions auto-republish to{ns}.deadletter.{path}.
Usage
from datetime import timedelta
from threetears.nats import NatsClient, Subjects
nc = await NatsClient.connect(
nats_url="nats://localhost:4222",
nats_subject_namespace="myapp",
client_name="my-service",
)
# Typed publish
await nc.publish(
subject=Subjects.audit_event("workspace.doc_set"),
message=AuditEvent(...),
)
# Typed request/reply
response = await nc.request(
subject=Subjects.tools_call(),
message=ToolCallRequest(...),
response_type=ToolCallResponse,
timeout=timedelta(seconds=5),
)
# Subscribe with Pydantic validation
sub = await nc.subscribe_typed(
subject=Subjects.audit_wildcard(),
cb=on_audit_event,
message_type=AuditEvent,
queue="audit-consumer",
)
# JetStream KV
bucket = await nc.kv_bucket(name="agent_config", ttl=timedelta(hours=2))
await bucket.put(key="agent-1", value=b"config-payload")
await nc.shutdown()
Distributed locks
The cross-pod job lock is not in this package: nats_distributed_lock (with LockHeld, LockHold, LockLost) lives in core, threetears.core.coordination, beside the KVLease it runs on. It takes the NatsClient from this package as its client. Only LockLossReason, the reason a lost hold reports, is defined here (threetears.nats.errors).
Object Store
NatsObjectStore wraps one Object Store bucket. Callers reach it through the client, never through js.object_store:
NatsClient.ensure_object_store(name=..., max_bytes=...)declares the bucket (memory storage by default) and is the bucket owner's call only.NatsClient.object_store(name=...)binds a bucket another identity declared and never creates one; a bucket nobody declared raisesObjectStoreNotFoundError.
On a bound store: put(name, data) -> ObjectInfo, get(name) -> bytes, info(name), list_objects(prefix=...) and bytes_held(). What it holds to:
- A name is written once. A second
putof a name raisesObjectExistsError; the server refuses it, so a racing writer cannot overwrite the first. A changed object is a new name. getchecks what it read against the object's recorded size and digest.- Reads use named consumers and the direct get, the forms a pod's grant admits; nats-py's own Object Store reads through an unnamed consumer a pod is refused.
- Deletion is the declarer's.
delete(name)andpurge_orphan_chunks(pointers=...)purge stream subjects, a management verb the hub holds and no pod does. The sweep removes chunks no object names only while no write claim stands in the pod's pointer bucket, judged by state and never by age. - A bucket at its
max_bytesrefuses a write withObjectStoreFullError.
A tool pod's own bucket is declared and swept by the hub on its behalf: threetears.nats.object_store_requests holds that contract (bind_pod_object_store, declare_pod_object_store, retire_pod_objects, and the request and reply models the hub imports).
KV errors
A KV failure the wrapper raises is a KvError, except the two refusals in the table, which are deliberately not: the L2 accessors catch KvError and degrade, and a refusal must stop the process instead. One KvError has its own type, because it needs a different response from the rest:
| Raised | Means | Respond by |
|---|---|---|
KvBucketNotFoundError (a KvError; .bucket names it) |
The server answered that the bucket's stream does not exist. | Waiting for the declarer, which has not declared it yet or not since a NATS restart wiped it. |
any other KvError |
The call failed for another reason. A request this principal is not granted is never answered, so a missing grant and an unreachable broker both arrive here as a deadline. | Checking the grant the message names, or the broker. |
KvConfigMismatch (NOT a KvError) |
A bind-only open found the live bucket carrying a configuration it refuses. | Running the declarer, which reconciles it. A bucket-wide expiry other than the opener's lifetime is reconciled only by a declarer that owns the bucket: ensure_kv_bucket(ttl=None, owns_bucket=True) from the bucket's one declarer. A bucket found live on file storage is recreated on memory, dropping its entries, only with the second opt-in drop_file_storage=True; without it the owner raises this same error and leaves the bucket untouched. |
StreamSubjectsOverlapError (NOT a KvError) |
A declaring open, or an operation recreating its bucket, found another stream owning the bucket's subjects. | Removing the conflicting stream, usually a client on the wrong subject namespace. |
KvBucketNotFoundError is raised by a bind-only open (kv_bucket / ensure_kv_bucket with create_if_missing=False) once its wait for the declarer is spent, by a declaring open whose create went unanswered and whose bind found nothing, and by an operation on a bound handle -- list_keys included -- whose stream vanished and could not be bound again. An existing except KvError still catches it.
A consumer holding a RAW nats-py handle (a KeyValue reached through jetstream_context()) gets nats-py's own exceptions. Classify them with the predicates instead of importing nats-py:
from threetears.nats import is_bucket_not_found, is_key_not_found, is_nats_error
try:
entry = await raw_kv.get(key)
except Exception as exc:
if is_key_not_found(exc):
entry = None # the bucket exists; the key does not
elif is_bucket_not_found(exc):
... # the bucket's stream is gone: wait for, or ask, its declarer
elif is_nats_error(exc):
... # any other failure the bus reported
else:
raise
is_bucket_not_found is also true of KvBucketNotFoundError, so one predicate serves a consumer holding both kinds of handle. None of the three reads message text.
Enforcement
Direct from nats import / from nats.aio imports are flagged by the per-repo enforcement walker tests/enforcement/test_nats_wrapper_usage.py. Strict mode by default; exemptions require a # rationale: ... line.
1 maintainer
Metadata
1 maintainer
Release files for 3tears-nats 0.69.0
For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.
Source distribution (sdist)
| File | Size | Uploaded | |
|---|---|---|---|
| 3tears_nats-0.69.0.tar.gz | 629.0 kB | Details |
Built distribution (wheel)
| File | Interpreter | ABI | Platform | Reset |
|---|---|---|---|---|
| 3tears_nats-0.69.0-py3-none-any.whl | Python 3 | none | any | Details |
Total release size: 970.0 kB
Release files / 3tears_nats-0.69.0.tar.gz
| Download URL | 3tears_nats-0.69.0.tar.gz |
|---|---|
| Size | 629.0 kB |
| Tags | Source |
|
SHA-256 checksum How to use checksums |
492fcdf899589e2e294d3dcd7be402121f8e72c5d69af89f33091cd352a6584e
|
|
BLAKE2b-256 checksum How to use checksums |
96e8e193cd39b422b97833dd9aafc7119db8c55ba518ac8f0feccdc69d2b27cb
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 16, 2026.
Transparency logRelease files / 3tears_nats-0.69.0-py3-none-any.whl
| Download URL | 3tears_nats-0.69.0-py3-none-any.whl |
|---|---|
| Size | 341.0 kB |
| Tags | Python 3 |
|
SHA-256 checksum How to use checksums |
504486d92f6d111b7f2c9652bd5584caffb938305b24ce03d8433bde6887a7f9
|
|
BLAKE2b-256 checksum How to use checksums |
f9da22330c97002486227b45b2262f0b11b507834f39905982231b96d6c25bcf
|
| Upload date | |
|
Uploaded using Trusted Publishing? What is trusted publishing? |
Yes |
| Uploaded via |
twine/7.0.0 CPython/3.13.14
|
Provenance
Provenance describes where a file came from. On PyPI, provenance is shared via attestations, which provide a verifiable record of the build or publishing details. View details, limitations and caveats.
PyPI Publish Attestation
PyPI verified that this artifact, at this checksum, originated from the publisher listed below.
Signed by GitHub Actions, verified by PyPI on Oct 16, 2026.
Transparency log
