Metadata-Version: 2.5
Name: greyhorse-nats
Version: 0.5.5
Summary: Greyhorse NATS library
Project-URL: Homepage, https://gitlab.com/max-plutonium/greyhorse
Project-URL: Repository, https://gitlab.com/max-plutonium/greyhorse
Author-email: Max Plutonium <plutonium.max@gmail.com>
Maintainer-email: Max Plutonium <plutonium.max@gmail.com>
License-Expression: MIT
License-File: LICENSE
Keywords: greyhorse,jetstream,messaging,nats
Classifier: Development Status :: 4 - Beta
Classifier: Framework :: AsyncIO
Classifier: Intended Audience :: Developers
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3 :: Only
Classifier: Programming Language :: Python :: 3.14
Classifier: Topic :: Software Development :: Libraries
Classifier: Topic :: System :: Distributed Computing
Classifier: Topic :: System :: Networking
Classifier: Typing :: Typed
Requires-Python: >=3.14
Requires-Dist: greyhorse~=0.5.5
Requires-Dist: nats-py~=2.15.0
Requires-Dist: pydantic-settings~=2.14.2
Provides-Extra: nkeys
Requires-Dist: nats-py[nkeys]~=2.15.0; extra == 'nkeys'
Description-Content-Type: text/markdown

Greyhorse NATS library
=======================

Greyhorse framework library for NATS/JetStream support (async only --
nats-py has no sync client).

The primary API is the **pieces**, not a ready-made module:

| Piece | Role |
|---|---|
| `NatsFragment` | material -- builds the engine |
| `NatsBorder` | lifecycle -- start/stop plus a real RTT liveness probe |
| `NatsAccess` | access -- hands out a connection window and a JetStream window |
| `NatsEngine` | the resource itself |
| `NatsModule` | ready-made single-connection floor, sugar over the three pieces above |

An application lists the ones it needs on its own `Module`, alongside
pieces from any other storage/messaging library -- no subclassing, no
multiple inheritance. See `examples/`.

`NatsStreams`/`NatsKv`/`NatsObjects` are a separate, second layer: plain,
stateless facades a consumer builds by hand over a live `JetStreamContext`
(reached through `NatsJsCtx`). They add nothing to the DI graph -- their only
job is turning nats-py exceptions into `Result[T, NatsError]`, so nothing
raises for an *expected* failure (a missing stream/bucket/key/object, a
timed-out request, a broker that is down).


How to build
------------

- Install the project

    ``uv python pin 3.14``

    ``uv venv``

    ``uv sync``

    ``source .venv/bin/activate``

- Format code commands

    ``ruff check --unsafe-fixes --fix``

    ``ruff format``

    ``mypy greyhorse_nats``

- Run tests

    ``pytest``


Usage
-----

Every snippet below is a runnable program, checked against a live broker. The
longer, commented versions live in `examples/` and are executed by the test
suite, so they cannot rot silently.

### One engine, one connection

`NatsModule` is the ready-made bundle for the single-broker case. The config
reaches the engine's constructor through the same `args={Type: value}` door
every `greyhorse.strand` resource uses -- there is no nats-specific wiring.

```python
import asyncio

from greyhorse.run import wrap_async
from greyhorse.strand import running
from greyhorse_nats import NatsConf, NatsConnCtx, NatsModule


async def main() -> None:
    conf = NatsConf(servers='nats://localhost:4222')

    with running(NatsModule, args={NatsConf: conf}) as module:
        conn_ctx = module.get(NatsConnCtx).unwrap()
        async with conn_ctx as nc:
            sub = await nc.subscribe('greet')
            await nc.publish('greet', b'hello')
            msg = await sub.next_msg(timeout=2)
            print(msg.data.decode())
            await sub.unsubscribe()


asyncio.run(wrap_async(main))
```

The engine connects when the module starts and drains when it stops; `nc` is
an ordinary `nats.aio.client.Client` borrowed for the length of the `async
with` block. See `examples/01_minimal.py`.

**Always drive the program through `greyhorse.run.wrap_async` (or `run`),
never a bare `asyncio.run(main())` around application code.** A NATS client
and its subscriptions belong to the loop they were created on, and the
border's `invoke_sync` bridge has to land on that same loop. Skip the wrapper
and the module comes up, works, and then fails at shutdown with `Event loop
is closed` -- an error the package recognises and explains, but one you never
need to see.

### Request/reply

Same connection, both roles. `request()` needs a live subscriber, so flush
before you rely on one being registered server-side.

```python
async with conn_ctx as nc:
    sub = await nc.subscribe('rpc.echo')
    await nc.flush()  # the server now knows about the subscription

    async def responder() -> None:
        request = await sub.next_msg(timeout=5)
        await request.respond(b'pong')

    task = asyncio.create_task(responder())
    reply = await nc.request('rpc.echo', b'ping', timeout=5)
    await task
    print(reply.data.decode())
    await sub.unsubscribe()
```

### JetStream

`NatsJsCtx` is the second product of the same engine -- a `JetStreamContext`
over the same connection, not a second one. `NatsStreams` is a plain facade
over it: every method answers with `Result`, so a missing stream or a timed
out fetch is a value, not an exception.

```python
from greyhorse_nats import NatsJsCtx, NatsStreams

async with module.get(NatsJsCtx).unwrap() as js:
    streams = NatsStreams(js)

    await streams.ensure_stream('ORDERS', ['orders.>'])
    ack = await streams.publish('orders.new', b'order-1')
    print(ack.unwrap().stream, ack.unwrap().seq)

    pulled = await streams.pull('orders.new', durable='worker', stream='ORDERS')
    fetched = await streams.fetch(pulled.unwrap(), batch=1, timeout=5)
    for msg in fetched.unwrap():
        print(msg.data.decode())
        await msg.ack()

    missing = await streams.stream_info('NO_SUCH_STREAM')
    print(missing.is_err())  # True -- NatsError.NotFound, nothing raised
```

An idle `fetch` that simply had nothing to deliver is `Ok([])`, while a
broker that stopped answering is `Err(NatsError.Unavailable)` -- the two look
identical at the nats-py level, and telling them apart is most of why this
facade exists. See `examples/03_jetstream.py`.

### Key-value and object storage

```python
from greyhorse_nats import NatsKv, NatsObjects

async with module.get(NatsJsCtx).unwrap() as js:
    kv = NatsKv(js, 'settings')
    await kv.ensure()
    await kv.put('greeting', b'hello kv')
    print((await kv.get('greeting')).unwrap().value.decode())
    print((await kv.get('no-such-key')).is_err())  # True

    objects = NatsObjects(js, 'reports')
    await objects.ensure()
    await objects.put('daily.txt', b'contents')
    print((await objects.get('daily.txt')).unwrap().data.decode())
```

A facade binds a bucket by NAME and remembers when the server says that
bucket is gone: after a delete it refuses until you call `ensure()` again,
rather than quietly recreating what an administrator removed. The limits of
that promise are spelled out under *JetStream facades* below. See
`examples/04_kv_objects.py`.

### A consumer that knows nothing about greyhorse

The class that uses NATS takes a context by TYPE. `NatsConnCtx` is an alias
for `AsyncShared[Client]` -- no marker, no base class, no import from this
package in the domain code:

```python
class PingService:
    def __init__(self, conn: NatsConnCtx) -> None:
        self._conn = conn

    async def ping(self) -> str:
        async with self._conn as nc:
            reply = await nc.request('rpc.echo', b'ping', timeout=5)
            return reply.data.decode()
```

The declaration is the only place that names this library, and it says what
the component takes and what it hands out:

```python
class PingComponent(Component):
    imports: ClassVar = Shared[NatsEngine]
    providers: ClassVar = NatsAccess
    exports: ClassVar = PingService
```

Two components declaring `providers = NatsAccess` over the SAME engine each
get their own provider instance and their own borrow -- one connection,
several independent consumers. See `examples/02_component.py`, and
`examples/05_multi_storage.py` for two engines on one floor without
subclassing anything.


Configuration from the environment
-----------------------------------

`NatsConf` above is built in code. A deployed service usually wants it from
the environment instead -- that is `NatsSettings`, a `pydantic_settings`
model reading `NATS_`-prefixed variables (and a `.env` file):

```python
from greyhorse_nats import NatsSettings

conf = NatsSettings().engine_conf()   # NATS_HOST, NATS_PORT, NATS_USER, ...
conf = NatsSettings().engine_conf(connect_timeout_seconds=5)   # tune anything
```

`engine_conf()` is the only supported way to turn settings into a config.
`settings.servers` carries **broker locations only** -- no credentials live in
a DSN any more -- so hand-building `NatsConf(servers=settings.servers, ...)`
silently drops every credential variable -- `NATS_USER`, `NATS_PASSWORD`,
`NATS_PASSWORD_FILE`, `NATS_TOKEN`, `NATS_CREDENTIALS_FILE`,
`NATS_NKEYS_SEED_FILE` -- and connects unauthenticated. Keyword arguments pass through to
`NatsConf` for the tuning knobs that have no env var of their own (timeouts,
pool sizes, `no_echo`).

| variable | default | notes |
|---|---|---|
| `NATS_SERVERS` | assembled from host/port | JSON list, bare DSN, or multi-host DSN -- see below |
| `NATS_HOST` / `NATS_PORT` | `localhost` / `4222` | used only when `NATS_SERVERS` is unset |
| `NATS_SCHEME` | `nats` | e.g. `tls` |
| `NATS_NAME` | `greyhorse` | the client name the broker reports |
| `NATS_USER` / `NATS_PASSWORD` | unset | carried as config FIELDS, never inside the DSN |
| `NATS_PASSWORD_FILE` | unset | read from disk, so the secret never sits in an env var |
| `NATS_TOKEN` | unset | alternative to user/password |
| `NATS_CREDENTIALS_FILE` | unset | nkeys/JWT creds file; needs the `nkeys` extra |
| `NATS_NKEYS_SEED_FILE` | unset | nkeys seed file; needs the `nkeys` extra |

`NATS_SERVERS` takes any of three shapes:

| value | result |
|---|---|
| `["nats://a:4222","nats://b:4333"]` | two servers (the JSON list pydantic imposes on a list-typed setting) |
| `nats://a:4222` | one server -- a bare DSN needs no JSON |
| `nats://a:4222,b:4333` | ONE multi-host DSN, NATS's own cluster spelling |
| empty or unset | falls back to `NATS_HOST`/`NATS_PORT` |

A DSN naming no host (`nats://`, or `nats://$BROKER` with the variable
unset) is **refused**, not quietly turned into `nats://localhost:4222` the
way pydantic would have it -- otherwise a templating slip points the service
at whatever broker happens to run locally, and nothing looks wrong.

`NATS_PASSWORD_FILE` exists for the deployment shape where secrets arrive as
mounted files (Kubernetes, Docker secrets) rather than environment variables.
Setting it makes the file **mandatory**: if it is missing, unreadable or
empty, `engine_conf()` raises instead of quietly connecting unauthenticated,
so a missing mount fails at configuration time rather than surfacing later as
a server-side auth error.

**Credentials never live in a DSN.** `NatsConf.servers` refuses one that
carries userinfo, because a DSN is a plain value pydantic serializes verbatim
from `model_dump()`/`model_dump_json()` -- no amount of `repr` work reaches
those. Pass credentials as fields instead:

```python
NatsConf(servers='nats://broker:4222', user='alice', password='...')   # user/password
NatsConf(servers='nats://broker:4222', token='...')                    # token
NatsConf(servers='nats://broker:4222', credentials_file='/run/secrets/app.creds')  # needs [nkeys]
```

**Exactly one auth mode per config**, and the model enforces it. The modes are
not additive: nats-py picks one through an `elif` chain, so a config carrying
two would have the loser silently dropped -- and the loser is not always the
same one, because the nkeys branch is taken only when the server offers a
nonce. Rather than ship a credential whose effect depends on which server
answered, these are rejected when the config is built:

| refused | because |
|---|---|
| `user` without `password`, or `password` without `user` | NATS authenticates with the pair; neither half is sent alone |
| `token` with a **complete** `user`+`password` pair | user/password wins, the token is dropped |
| `credentials_file` with `nkeys_seed_file` | the seed is never read |
| `credentials_file`/`nkeys_seed_file` with `user`/`password` | which one wins depends on the server |
| `NATS_TOKEN` with `NATS_USER` + `NATS_PASSWORD_FILE` † | same clash, one step earlier |
| `NATS_PASSWORD_FILE` without `NATS_USER` † | the resolved password would never be sent |
| `NATS_PASSWORD_FILE` with `NATS_CREDENTIALS_FILE`/`NATS_NKEYS_SEED_FILE` † | same clash as `password`, one step earlier |

A token beside HALF a pair is **not** the token clash: there nats-py sends
the token and nothing else, so the incomplete pair is the defect, and that is
what the error names. Saying "your token is ignored" would send you to delete
the one credential that works.

An empty value is **not** a value: `NATS_TOKEN=''` (or `'   '`) means unset,
not "authenticate with an empty token". Without that, a blank variable from
an unfilled template connected anonymously and looked configured.

† `NatsSettings` only — `NatsConf` has no `password_file` field, so these
three are caught by `engine_conf()` before the file is opened, and reported as
the mode clash they are rather than as a file it could not read.

A `token` **alongside** `credentials_file`/`nkeys_seed_file` is supported --
nats-py sends both, for auth callouts.

`credentials_file` and `nkeys_seed_file` need the signing support, which is an
optional extra: `pip install greyhorse-nats[nkeys]`. Without it the config is
refused with that instruction, rather than failing later inside `nats.connect()`
with a bare `ModuleNotFoundError`.

`password` and `token` are `SecretStr`, so they render as `**********` in
every repr, dump and traceback and yield their real value only where the
connection is actually made. A bare username in a URL counts as a credential
too -- nats-py reads it as an **auth token**, not as a user name.

Not to be confused with `NATS_TEST_URI` below: that one is a *test-suite* gate
and has nothing to do with `NatsSettings`.

### Who owns reconnect

An outage can be repaired two ways, and the package supports both without
choosing between them: the nats-py driver's own reconnect loop
(`allow_reconnect`, `max_reconnect_attempts`, `reconnect_time_wait_seconds` on
`NatsConf` -- `True`/`60`/`2` by default), or the border's repair
(`NatsBorder.check()` drives `NatsEngine.is_alive()`, a real RTT probe; a
failing check feeds light repair at the border and, on a permanent failure,
heavy repair at the controller). With the default `allow_reconnect=True` the
two genuinely overlap: the driver may already be reconnecting underneath the
same connection while `check()` still reports the broker unreachable.

Keep the default. Driver-level reconnect is cheaper -- it resumes the same
connection and its subscriptions rather than tearing the engine down -- and
the border's repair still stands as the outer net for the case the driver
gives up (`max_reconnect_attempts` exhausted). Setting `allow_reconnect=False`
hands the whole job to the border instead, and turns every outage, however
short, into a full rebuild of the engine; a consumer who wants that path can
additionally bound the controller's own escalation with a
`ControllerRestartPolicy` on the resource (`Resource(NatsEngine,
operators=NatsBorder, policy=ControllerRestartPolicy(...))`). A consumer who
does not want to think about it should leave `allow_reconnect` at its default
of `True`.


JetStream facades
------------------

```python
js_ctx = module.get(NatsJsCtx).unwrap()
async with js_ctx as js:
    streams = NatsStreams(js)
    match await streams.ensure_stream('ORDERS', ['orders.>']):
        case Ok(info):
            ...
        case Err(err):
            ...  # NatsError -- never an exception
```

See `examples/03_jetstream.py` and `examples/04_kv_objects.py`.

**A KV/object facade is bound to a bucket NAME, not to a bucket.** nats-py's
handle carries the name and the JetStream context; nothing in it refers to
the bucket instance. Two consequences worth knowing before you rely on one:

* A bucket deleted **and recreated** under the same name is adopted silently
  by an existing facade -- ordinary calls keep succeeding, against the new
  bucket. Only `NatsObjects.list()` catches it, because that path already
  asks the server and can compare identity for free. Checking everywhere
  would mean a round-trip on every `put` and every `get`, which is not a
  price this library pays. After an administrative delete, build a new
  facade rather than trusting an old one.
* A bucket deleted and **left** deleted is remembered, and the facade then
  refuses until you call `ensure()` deliberately -- but only the calls that
  can prove it do the remembering. `keys`, `history`, `put`, `purge` and
  `list` surface a bucket-level error and latch; `get` (and `info`/`delete`
  on the object side) raise exactly what an ordinary missing key or object
  raises, so they report `NotFound` correctly and record nothing.


Tests
-----

Most of the suite (config parsing, `plan()`-only DI checks) needs no server
at all. The rest needs a running NATS broker with JetStream enabled:

    ``docker compose -f tests/docker-compose.yml up -d --wait``

They are gated behind the `NATS_TEST_URI` environment variable; without it
they skip with a clear reason instead of erroring on "connection refused".

    ``NATS_TEST_URI=nats://localhost:54222 pytest``

`tests/docker-compose.yml` brings up the broker on non-default ports
(`54222`/`58222`, container name `greyhorse-nats-throwaway`) so it never
collides with a NATS instance already running on the standard `4222`/`8222`.
Both host ports are overridable -- `NATS_CLIENT_PORT` and `NATS_MONITOR_PORT`
-- for the case those two are taken as well.

Afterwards, take the broker down and its volume with it:

    ``docker compose -f tests/docker-compose.yml down -v``
