diff --git a/.github/workflows/ci.yml b/.github/workflows/ci.yml index e8d5fdc..b396ac3 100644 --- a/.github/workflows/ci.yml +++ b/.github/workflows/ci.yml @@ -48,7 +48,7 @@ jobs: uv venv --python 3.14 .venv-minimal VIRTUAL_ENV=.venv-minimal uv pip install . VIRTUAL_ENV=.venv-minimal uv run --no-project python -c " - import kit, kit.health, kit.httpapi, kit.config, kit.observability, kit.email + import kit, kit.health, kit.httpapi, kit.config, kit.observability, kit.email, kit.messaging print('kit', kit.__version__, 'imports without the otel extra') " diff --git a/README.md b/README.md index 67f5572..9b1eaf9 100644 --- a/README.md +++ b/README.md @@ -24,6 +24,7 @@ from kit.health import Registry | `kit.clients` | Outbound calls: deadlines, retries with jitter, a circuit breaker, trace and request-id propagation, pluggable auth | | `kit.config` | Env loading, the service prefix, the `PORT` and `OTEL_*` exceptions, fail-fast process settings and fail-soft backing systems | | `kit.email` | Outbound mail: the message every provider agrees on, and one Resend adapter on top of `kit.clients` | +| `kit.messaging` | The event bus: publish and pull-subscribe over NATS JetStream, explicit ack, nak and term, and the four headers tenancy and idempotency travel in | | `kit.testing` | The contract tests a service inherits, so the shared behaviour is verified in every repository | About a thousand lines. The most important piece is also the smallest: roughly @@ -95,6 +96,20 @@ honest is what it refuses to hold. Templates, rendered copy and which person receives which message stay in the service, because those are the parts that differ, and a package holding them would be a mail product rather than a seam. +`kit.messaging` is admitted on the same argument and is worth checking against +it just as hard, because it arrived on its FIRST copy rather than its third. It +is not an abstraction over a domain: it is the ack policy, the redelivery +behaviour, the reconnection settings and the header convention, and those go +wrong the same way in every service that decides them alone. They also go wrong +invisibly, and are found during an incident, with a consumer that has been +quietly reprocessing or quietly skipping. + +The boundary that keeps it honest is again what it refuses to hold. The payload +is opaque bytes: there is no schema, no event registry and no versioning, because +what an event IS belongs to the services that agree on it. It creates no streams +and no consumers either, since retention and replica counts are operational +decisions, and a library that makes them makes them once, wrongly, in production. + Business models, ORM models, migrations, route trees and service settings never belong here at all. diff --git a/pyproject.toml b/pyproject.toml index 8bd0afe..8972a51 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ [project] name = "scadable-kit" -version = "0.6.1" +version = "0.7.0" description = "Cross-cutting behaviour shared by every SCADABLE service" requires-python = ">=3.14,<3.15" license = { file = "LICENSE" } @@ -45,6 +45,12 @@ otel = [ # traceparent, B starts a new trace, and one request is two unlinked traces. "opentelemetry-instrumentation-httpx>=0.65b0", ] +# The bus is an extra for the same reason telemetry is: most services never +# touch it, and a dependency here is one every service in the fleet inherits and +# cannot refuse. `kit.messaging` imports cleanly without it, and raises a +# BrokerUnavailable naming this extra if something actually tries to connect, +# which is a failure a service that forgot can read. +nats = ["nats-py>=2.15,<3"] [build-system] requires = ["hatchling"] diff --git a/src/kit/__init__.py b/src/kit/__init__.py index 1e3d33d..844a8f0 100644 --- a/src/kit/__init__.py +++ b/src/kit/__init__.py @@ -20,4 +20,4 @@ across the fleet without opening every repository. """ -__version__ = "0.6.1" +__version__ = "0.7.0" diff --git a/src/kit/messaging/__init__.py b/src/kit/messaging/__init__.py new file mode 100644 index 0000000..118bb63 --- /dev/null +++ b/src/kit/messaging/__init__.py @@ -0,0 +1,88 @@ +"""The event bus, as transport policy the whole fleet shares. + + bus = MessageBus(Broker(name="events", servers=("nats://nats.nats:4222",))) + await bus.connect() + + await bus.publish( + "events.repository.indexed", + payload, + Envelope(tenant=tenant, event_id=uid, event_type="repository.indexed", + occurred_at=now), + ) + + subscription = await bus.subscribe(stream="EVENTS", subject="events.>", + durable="brain") + for message in await subscription.fetch(batch=8): + await handle(message) + await message.ack() + +WHY THIS IS IN THE KIT AT ALL, given that "code moves in on its third copy". This +is the `kit.clients` case rather than the database-helpers case: it is not an +abstraction over a domain, it is the ack policy, the redelivery behaviour, the +reconnection settings and the header convention, and those are wrong in the same +way in every service that decides them alone. They are also wrong in a way that +only shows up during an incident, when a consumer has been quietly reprocessing +or quietly skipping. + +THE BOUNDARY THAT KEEPS IT HONEST IS WHAT IT REFUSES TO HOLD. The payload is +opaque bytes. There is no schema, no event registry and no versioning: what an +event IS belongs to the services that agree on it, and a package holding that +would be a messaging product rather than a seam. It creates no streams and no +consumers either, because retention and replica counts are operational decisions +and a library that makes them makes them once, wrongly, in production. + +WHAT IS POLICY AND WHAT IS NOT. Timeouts, batch sizes, reconnection and nak +delays are parameters and yours to change per broker. The four header names are +not: they are how tenancy and idempotency travel, and a service that renames one +stops being readable by every other service on the bus. + +AT LEAST ONCE, SAID PLAINLY. `fetch` returns unsettled messages and settling is +an explicit `ack`, `nak` or `term`. Nothing here acknowledges on your behalf, so +a handler that raises gets its message back rather than losing it. Handlers must +be idempotent; `Envelope.event_id` is what they deduplicate on, and it rides the +wire as `Nats-Msg-Id` so the server suppresses duplicate publishes too. +""" + +from __future__ import annotations + +from kit.messaging._broker import ( + DEFAULT_FETCH_TIMEOUT_SECONDS, + Broker, + Message, + MessageBus, + Subscription, +) +from kit.messaging._errors import ( + BrokerError, + BrokerTimeout, + BrokerUnavailable, + MalformedMessage, +) +from kit.messaging._headers import ( + EVENT_ID, + EVENT_TYPE, + OCCURRED_AT, + TENANT, + Envelope, + envelope, +) +from kit.messaging._health import register_brokers + +__all__ = [ + "DEFAULT_FETCH_TIMEOUT_SECONDS", + "EVENT_ID", + "EVENT_TYPE", + "OCCURRED_AT", + "TENANT", + "Broker", + "BrokerError", + "BrokerTimeout", + "BrokerUnavailable", + "Envelope", + "MalformedMessage", + "Message", + "MessageBus", + "Subscription", + "envelope", + "register_brokers", +] diff --git a/src/kit/messaging/_broker.py b/src/kit/messaging/_broker.py new file mode 100644 index 0000000..eb6116d --- /dev/null +++ b/src/kit/messaging/_broker.py @@ -0,0 +1,383 @@ +"""The bus, as one object per broker, built where everything else is built. + +WHAT THIS OWNS. Connecting, publishing with the four headers, pulling a batch, +and the three ways a message can be settled. Reconnection is delegated to the +NATS client, whose parameters are set here rather than left at their defaults. + +WHAT IT REFUSES TO OWN, and this is the important half: + + - IT CREATES NO STREAMS AND NO CONSUMERS. Creating a stream carries retention, + replica and discard decisions, and a library that creates one on connect + eventually creates the wrong one in production, silently, because the call + is idempotent and the second caller's arguments are ignored. Streams are an + operational act. + + - IT DOES NOT AUTO-ACK. `fetch` hands back messages that are still unsettled, + and settling is an explicit call. An auto-acking API turns at-least-once + into at-most-once at the moment a handler raises, which is the one moment + anybody cares. + + - IT HOLDS NO SCHEMA. The payload is opaque bytes. + +THE IMPORT OF `nats` IS FUNCTION LOCAL, always, because the dependency is an +optional extra and every service in the fleet inherits this package whether or +not it touches a bus. A module-scope import would make `import kit.messaging` +fail at startup for the services that never asked for it. +""" + +from __future__ import annotations + +import logging +from collections.abc import Sequence +from dataclasses import dataclass, field +from typing import Any + +from kit.messaging._errors import BrokerTimeout, BrokerUnavailable, MalformedMessage +from kit.messaging._headers import Envelope, envelope +from kit.observability import ( + MESSAGING_CONSUMED, + MESSAGING_DELIVERIES, + MESSAGING_PUBLISHED, + MESSAGING_SETTLED, + record, +) + +log = logging.getLogger("kit.messaging") + +DEFAULT_CONNECT_TIMEOUT_SECONDS = 5.0 +DEFAULT_RECONNECT_WAIT_SECONDS = 2.0 +DEFAULT_FETCH_TIMEOUT_SECONDS = 5.0 +DEFAULT_PUBLISH_TIMEOUT_SECONDS = 5.0 + +RECONNECT_FOREVER = -1 +"""What `max_reconnect_attempts` means by default, and it is deliberate. + +A consumer that gives up reconnecting becomes a process that is running, healthy +by any liveness probe, and consuming nothing. Retrying forever turns that into a +visible gap in throughput instead, which is the failure somebody notices. +""" + + +@dataclass(frozen=True, slots=True) +class Broker: + """One bus, and everything this service knows about reaching it.""" + + name: str + """Short and STABLE. It becomes a metric label and a readiness check name, so + changing it starts a new time series and orphans the dashboard watching the + old one.""" + + servers: tuple[str, ...] = () + credentials_path: str = "" + """A `.creds` file, when the bus requires one. + + Empty today: the deployed bus has no authentication and is reached only + through a NetworkPolicy. It is carried anyway so that turning auth on is a + config change in one service rather than a release of this package.""" + + token: str = "" + connect_timeout_seconds: float = DEFAULT_CONNECT_TIMEOUT_SECONDS + reconnect_wait_seconds: float = DEFAULT_RECONNECT_WAIT_SECONDS + max_reconnect_attempts: int = RECONNECT_FOREVER + publish_timeout_seconds: float = DEFAULT_PUBLISH_TIMEOUT_SECONDS + + def __post_init__(self) -> None: + if not self.name.strip(): + raise ValueError("a broker needs a name; it labels metrics and health checks") + + @property + def configured(self) -> bool: + """Whether an operator set this up, answered without opening a socket.""" + return bool(self.servers) + + +@dataclass(frozen=True, slots=True) +class Message: + """One delivery, still unsettled. + + `delivered` is the attempt number this delivery is, counting from one. Above + one means the message has been redelivered, which is the signal a handler + uses to decide that something is poisonous rather than merely slow. + """ + + envelope: Envelope + payload: bytes + delivered: int + stream: str + _broker: str = field(repr=False, default="") + _message: Any = field(repr=False, default=None) + + async def ack(self) -> None: + """Done. Do not send this again.""" + await self._settle("ack") + + async def nak(self, delay_seconds: float | None = None) -> None: + """Not now. Send it again, optionally after a delay. + + The delay is the back pressure knob: naking without one on a dependency + that is down produces a hot loop against it. + """ + await self._settle("nak", delay_seconds) + + async def term(self) -> None: + """Never. Stop redelivering this message. + + For a message that will fail identically every time: a malformed + payload, or work whose subject no longer exists. A nak would retry it + until the consumer's limit and then dead-letter it anyway, having spent + the attempts to learn what was already known. + """ + await self._settle("term") + + async def _settle(self, outcome: str, delay_seconds: float | None = None) -> None: + try: + if outcome == "ack": + await self._message.ack() + elif outcome == "nak": + await self._message.nak(delay=delay_seconds) + else: + await self._message.term() + except Exception as error: + # A FAILED SETTLE IS NOT A FAILED HANDLER. The work happened; only + # the acknowledgement did not, so the message will be redelivered + # and the handler's idempotency is what saves it. Raising here would + # tell a caller its work failed, which is worse than the truth. + log.warning( + "could not settle message", + extra={"broker": self._broker, "outcome": outcome, "error": str(error)}, + ) + return + record(MESSAGING_SETTLED, 1, broker=self._broker, stream=self.stream, outcome=outcome) + + +class Subscription: + """A durable consumer, pulled in batches. + + PULL RATHER THAN PUSH, because a pull consumer takes work at the rate the + process can finish it. A push consumer is handed messages at the server's + rate, and a slow handler turns that into an ack timeout and redelivery of + work that is still running. + """ + + def __init__(self, broker: str, stream: str, subscription: Any) -> None: + self.broker = broker + self.stream = stream + self._subscription = subscription + + async def fetch( + self, batch: int = 1, timeout_seconds: float = DEFAULT_FETCH_TIMEOUT_SECONDS + ) -> Sequence[Message]: + """Take up to `batch` messages, or none. + + AN EMPTY BATCH IS NOT AN ERROR and is the ordinary case on a quiet + stream. The NATS client signals it with a timeout, which is caught here + and turned into an empty sequence, so a caller's loop does not have to + treat "nothing happened" as a failure. + """ + try: + raw = await self._subscription.fetch(batch, timeout=timeout_seconds) + except TimeoutError: + return () + except Exception as error: + raise BrokerUnavailable(self.broker, f"fetch failed: {error}") from None + + messages: list[Message] = [] + for one in raw: + try: + read = envelope(dict(one.headers or {})) + except ValueError as error: + # TERMINATED, NOT NAKED. It will be malformed next time too, and + # a nak would spend the whole attempt budget rediscovering that. + await _terminate(one) + log.warning( + "terminated an unreadable message", + extra={"broker": self.broker, "stream": self.stream, "error": str(error)}, + ) + continue + delivered = _delivery_count(one) + record(MESSAGING_CONSUMED, 1, broker=self.broker, stream=self.stream) + record(MESSAGING_DELIVERIES, delivered, broker=self.broker, stream=self.stream) + messages.append( + Message( + envelope=read, + payload=one.data, + delivered=delivered, + stream=self.stream, + _broker=self.broker, + _message=one, + ) + ) + return tuple(messages) + + +class MessageBus: + """The connection, built once in a composition root and passed around. + + Never a module-level singleton: a test substitutes one by passing a + different object, not by monkeypatching this module. + """ + + def __init__(self, broker: Broker, connection: Any | None = None) -> None: + """`connection` exists for tests, and for one real case. + + A test passes a stub and exercises the real header, settle and metric + code with no server and no sleep. The real case is a service that + already holds a connection and wants this package's policy on top of it. + """ + self.broker = broker + self._connection = connection + self._stream: Any = None if connection is None else connection.jetstream() + + @property + def connected(self) -> bool: + """What the client believes, without asking the network. + + A readiness check reads this. It does not ping the bus, for the reason + `kit.clients` does not call its upstreams: a probe that reaches out turns + one kubelet's polling interval into load on a shared dependency, from + every replica, forever. + """ + return self._connection is not None and bool(self._connection.is_connected) + + async def connect(self) -> None: + """Open the connection, or say why not. + + Idempotent: calling it twice is a no-op rather than a second connection, + because a composition root that retries should not leak one. + """ + if self._connection is not None: + return + if not self.broker.configured: + raise BrokerUnavailable(self.broker.name, "no servers configured") + + try: + import nats as _nats + except ImportError: + raise BrokerUnavailable( + self.broker.name, + "the nats extra is not installed; add scadable-kit[nats]", + ) from None + + options: dict[str, Any] = { + "servers": list(self.broker.servers), + "connect_timeout": self.broker.connect_timeout_seconds, + "reconnect_time_wait": self.broker.reconnect_wait_seconds, + "max_reconnect_attempts": self.broker.max_reconnect_attempts, + } + if self.broker.credentials_path: + options["user_credentials"] = self.broker.credentials_path + if self.broker.token: + options["token"] = self.broker.token + + # ERASED AT THE BOUNDARY, the way `_metrics` erases the OpenTelemetry + # SDK. `nats-py` ships no type information, so under pyright's strict + # mode every call through it is a partially-unknown type. Narrowing here + # keeps the strictness on OUR code, where it is worth having, rather + # than turning the setting off for the package. + nats: Any = _nats + + try: + self._connection = await nats.connect(**options) + except TimeoutError: + raise BrokerTimeout(self.broker.name, "connect timed out") from None + except Exception as error: + raise BrokerUnavailable(self.broker.name, f"connect failed: {error}") from None + self._stream = self._connection.jetstream() + + async def publish(self, subject: str, payload: bytes, message: Envelope) -> None: + """Publish, and wait for the stream to say it stored it. + + WAITING IS THE POINT. JetStream's publish returns an acknowledgement + that the message is persisted to the stream's replicas; not waiting for + it makes this a fire and forget call that returns success while the + message is still in flight, which is indistinguishable from working + right up until a leader election eats one. + + Deduplication comes free: the envelope's id rides as `Nats-Msg-Id`, and + the stream refuses a second copy inside its duplicate window. + """ + stream = self._require_stream() + try: + await stream.publish( + subject, + payload, + headers=message.headers(), + timeout=self.broker.publish_timeout_seconds, + ) + except TimeoutError: + raise BrokerTimeout(self.broker.name, "publish timed out") from None + except Exception as error: + raise BrokerUnavailable(self.broker.name, f"publish failed: {error}") from None + record(MESSAGING_PUBLISHED, 1, broker=self.broker.name, stream=subject.split(".")[0]) + + async def subscribe(self, stream: str, subject: str, durable: str) -> Subscription: + """Bind to a durable consumer that already exists. + + `durable` is a name the operator created. Binding rather than creating is + what keeps ack policy, max deliveries and the dead-letter arrangement in + one place instead of in whichever service connected first. + """ + pull = self._require_stream() + try: + subscription = await pull.pull_subscribe(subject, durable=durable, stream=stream) + except Exception as error: + raise BrokerUnavailable( + self.broker.name, f"could not bind consumer {durable!r}: {error}" + ) from None + return Subscription(self.broker.name, stream, subscription) + + async def close(self) -> None: + """Drain and close, so in-flight work finishes. + + `drain` rather than `close`: it stops taking new messages, lets what is + already in hand settle, and then closes. A bare close abandons them and + they come back as redeliveries to whoever is left. + """ + if self._connection is None: + return + try: + await self._connection.drain() + except Exception as error: + log.warning( + "could not drain the broker connection", + extra={"broker": self.broker.name, "error": str(error)}, + ) + self._connection = None + self._stream = None + + def _require_stream(self) -> Any: + if self._stream is None: + raise BrokerUnavailable(self.broker.name, "not connected") + return self._stream + + +async def _terminate(message: Any) -> None: + """Terminate a message we could not read, and never raise doing it.""" + try: + await message.term() + except Exception: + return + + +def _delivery_count(message: Any) -> int: + """Which attempt this is, counting from one. + + Absent metadata is reported as a first delivery rather than as zero: zero + would say "never delivered" about a message being held, and the number is + used to decide whether something is poisonous. + """ + metadata = getattr(message, "metadata", None) + delivered = getattr(metadata, "num_delivered", None) + if not isinstance(delivered, int) or delivered < 1: + return 1 + return delivered + + +__all__ = [ + "DEFAULT_FETCH_TIMEOUT_SECONDS", + "Broker", + "MalformedMessage", + "Message", + "MessageBus", + "Subscription", +] diff --git a/src/kit/messaging/_errors.py b/src/kit/messaging/_errors.py new file mode 100644 index 0000000..38b8bc3 --- /dev/null +++ b/src/kit/messaging/_errors.py @@ -0,0 +1,54 @@ +"""What a broker failure is, as types a caller can branch on. + +EVERY ONE OF THESE NAMES THE BROKER, for the reason `kit.clients._errors` gives +about upstreams: a fleet running more than one bus, or one service reading from +two, produces a log line that costs an incident responder the first ten minutes +working out which. + +THEY CARRY NO STATUS CODE AND NO SUBJECT. Which HTTP status a service renders a +broker failure as belongs to the API version doing the rendering. The subject is +left off deliberately too: it frequently carries a tenant id, and an exception +message is the one place that reliably reaches a log aggregator unredacted. +""" + +from __future__ import annotations + + +class BrokerError(Exception): + """Something went wrong talking to the bus.""" + + def __init__(self, broker: str, detail: str = "") -> None: + super().__init__(f"{broker}: {detail}" if detail else broker) + self.broker = broker + self.detail = detail + + +class BrokerUnavailable(BrokerError): + """The bus could not be reached, or would not take the message. + + Retrying LATER may work. This is the shape a publisher should let its own + retry policy see, and the shape a caller should never turn into a dropped + message: on this error the state change that produced the message has + usually already been committed. + """ + + +class BrokerTimeout(BrokerUnavailable): + """We gave up waiting. + + A subclass of Unavailable, deliberately, because to a caller deciding what to + do next it is the same class of problem. Separate because it is the one that + means we stopped rather than the bus said no, and the two lead to different + places when somebody goes looking. + """ + + +class MalformedMessage(BrokerError): + """A message arrived that this package will not hand on. + + Its headers do not carry the four fields a consumer needs, so there is no + tenant to scope to and no id to dedupe on. Not a subclass of Unavailable: + retrying changes nothing, because the message will be malformed next time + too. The consumer's move is to terminate it rather than nak it, which is why + the two settle verbs are separate. + """ diff --git a/src/kit/messaging/_headers.py b/src/kit/messaging/_headers.py new file mode 100644 index 0000000..7ce04a0 --- /dev/null +++ b/src/kit/messaging/_headers.py @@ -0,0 +1,114 @@ +"""What travels beside a message, and why these four and no others. + +THE PAYLOAD IS THE SERVICE'S BUSINESS AND THESE ARE NOT. A publisher and a +consumer have to agree on where the tenant lives and on what makes a message the +same message, before either can read a byte of the body. Left to each service, +those two answers diverge, and the divergence is invisible until a consumer is +silently processing another tenant's event or reprocessing one it already did. + +WHAT IS DELIBERATELY NOT HERE. There is no schema, no version field, no envelope +around the payload. The body is opaque bytes to this package. `kit` holds the +transport policy; what an event IS belongs to the services that agree on it. +""" + +from __future__ import annotations + +from collections.abc import Mapping +from dataclasses import dataclass +from datetime import UTC, datetime + +TENANT = "Scadable-Tenant" +"""Which tenant the message belongs to. + +Not optional and not defaulted. A consumer scopes its work by this, and a +message that arrives without one cannot be scoped to anything, so it is refused +here rather than defaulted to something that would read as legitimate later. +""" + +EVENT_ID = "Nats-Msg-Id" +"""What makes this message the same message. NOT one of our own header names. + +`Nats-Msg-Id` is JetStream's own deduplication header: the server refuses a +second publish carrying an id it has already seen inside the stream's duplicate +window. Using our own name here would leave that mechanism switched off and +duplicate suppression entirely to consumers. + +So publishing is idempotent within the window for free, and past the window it +is still the id a consumer dedupes on. That is why the constant is named for +what it means to us and valued for what NATS does with it. +""" + +EVENT_TYPE = "Scadable-Event" +"""What kind of thing happened, so a consumer can route without parsing a body.""" + +OCCURRED_AT = "Scadable-Occurred-At" +"""When it happened, RFC 3339 in UTC. + +When it HAPPENED, not when it was published or delivered. Those differ whenever +a relay was behind, and the difference is the thing anybody debugging a lag +actually wants. +""" + + +@dataclass(frozen=True, slots=True) +class Envelope: + """The four headers, as one object, checked once. + + Frozen because a message's identity should not change between being received + and being settled. + """ + + tenant: str + event_id: str + event_type: str + occurred_at: datetime + + def __post_init__(self) -> None: + """Refuse a blank rather than carry one. + + A blank tenant scopes to nothing, a blank id makes every message unique + and defeats deduplication, and a blank type gives a consumer nothing to + route on. Each is a bug at the publisher that would otherwise be found + by the consumer, hours later, as absent data. + """ + for name, value in ( + ("tenant", self.tenant), + ("event_id", self.event_id), + ("event_type", self.event_type), + ): + if not value.strip(): + raise ValueError(f"{name} is required and was blank") + if self.occurred_at.tzinfo is None: + raise ValueError("occurred_at must be timezone aware") + + def headers(self) -> dict[str, str]: + """The wire form.""" + return { + TENANT: self.tenant, + EVENT_ID: self.event_id, + EVENT_TYPE: self.event_type, + OCCURRED_AT: self.occurred_at.astimezone(UTC).isoformat(), + } + + +def envelope(headers: Mapping[str, str] | None) -> Envelope: + """Read the four headers back, or say which one was missing. + + Raises `ValueError`, which the broker turns into a typed error naming itself. + A message that cannot be read is not a message this package will hand to a + consumer with fields quietly defaulted. + """ + present = headers or {} + occurred = present.get(OCCURRED_AT, "") + try: + at = datetime.fromisoformat(occurred) + except ValueError: + raise ValueError(f"{OCCURRED_AT} is not an RFC 3339 timestamp") from None + # A naive timestamp is ambiguous rather than wrong, and reading it as UTC + # would be a guess that looks like data. `Envelope` refuses it below. + return Envelope( + tenant=present.get(TENANT, ""), + event_id=present.get(EVENT_ID, ""), + event_type=present.get(EVENT_TYPE, ""), + occurred_at=at, + ) diff --git a/src/kit/messaging/_health.py b/src/kit/messaging/_health.py new file mode 100644 index 0000000..2f3329a --- /dev/null +++ b/src/kit/messaging/_health.py @@ -0,0 +1,32 @@ +"""The bus, in `/readyz`, without ever touching the network. + +INFORMATIONAL, NEVER BLOCKING, and the reason is the same one `kit.clients` gives +about an open breaker. A web replica that cannot reach the bus can still serve +every read it owns. Failing readiness would take it out of rotation, which turns +one component's outage into two and does not reconnect anything. + +It reads what the client already believes rather than sending a ping, because a +readiness probe that reaches out turns one kubelet's polling interval into load +on a shared dependency, from every replica, forever. +""" + +from __future__ import annotations + +from collections.abc import Awaitable, Callable, Iterable + +from kit.health import Registry +from kit.messaging._broker import MessageBus + + +def register_brokers(registry: Registry, buses: Iterable[MessageBus]) -> None: + """Declare one check per bus, named for the broker.""" + for bus in buses: + registry.add_informational(bus.broker.name, _check(bus)) + + +def _check(bus: MessageBus) -> Callable[[], Awaitable[None]]: + async def check() -> None: + if not bus.connected: + raise RuntimeError(f"{bus.broker.name} is not connected") + + return check diff --git a/src/kit/observability/__init__.py b/src/kit/observability/__init__.py index 3278d64..ddb1c2e 100644 --- a/src/kit/observability/__init__.py +++ b/src/kit/observability/__init__.py @@ -23,6 +23,10 @@ CLIENT_BREAKER_OPEN, CLIENT_DURATION, CLIENT_REQUESTS, + MESSAGING_CONSUMED, + MESSAGING_DELIVERIES, + MESSAGING_PUBLISHED, + MESSAGING_SETTLED, SERVER_DURATION, SERVER_REQUESTS, record, @@ -41,6 +45,10 @@ "CLIENT_BREAKER_OPEN", "CLIENT_DURATION", "CLIENT_REQUESTS", + "MESSAGING_CONSUMED", + "MESSAGING_DELIVERIES", + "MESSAGING_PUBLISHED", + "MESSAGING_SETTLED", "SERVER_DURATION", "SERVER_REQUESTS", "JSONFormatter", diff --git a/src/kit/observability/_metrics.py b/src/kit/observability/_metrics.py index a915a4a..edd3b96 100644 --- a/src/kit/observability/_metrics.py +++ b/src/kit/observability/_metrics.py @@ -40,6 +40,14 @@ CLIENT_ATTEMPTS = "http.client.attempts" CLIENT_BREAKER_OPEN = "http.client.breaker_open" +# MESSAGING. The label is the STREAM, never the subject: a subject carries a +# tenant id in most useful designs, and the module docstring above is about +# exactly that turning one series into millions. +MESSAGING_PUBLISHED = "messaging.published" +MESSAGING_CONSUMED = "messaging.consumed" +MESSAGING_SETTLED = "messaging.settled" +MESSAGING_DELIVERIES = "messaging.deliveries" + def start_metrics( *, @@ -128,6 +136,22 @@ def _build(meter: Any) -> dict[str, Any]: CLIENT_BREAKER_OPEN: meter.create_up_down_counter( CLIENT_BREAKER_OPEN, unit="1", description="Upstreams currently being shed" ), + MESSAGING_PUBLISHED: meter.create_counter( + MESSAGING_PUBLISHED, unit="1", description="Messages published, by stream" + ), + MESSAGING_CONSUMED: meter.create_counter( + MESSAGING_CONSUMED, unit="1", description="Messages received, by stream" + ), + MESSAGING_SETTLED: meter.create_counter( + MESSAGING_SETTLED, + unit="1", + description="Messages settled, by outcome: ack, nak or term", + ), + MESSAGING_DELIVERIES: meter.create_histogram( + MESSAGING_DELIVERIES, + unit="1", + description="Delivery attempt this message is on. Above 1 is redelivery.", + ), } diff --git a/tests/test_messaging.py b/tests/test_messaging.py new file mode 100644 index 0000000..06a782c --- /dev/null +++ b/tests/test_messaging.py @@ -0,0 +1,623 @@ +"""The bus, exercised against a stub connection rather than a server. + +The stub is the same seam `service_client(transport=...)` uses: the real header, +settle, metric and error code runs, with no socket and no sleep of consequence. +""" + +from __future__ import annotations + +import builtins +import sys +import types +from datetime import UTC, datetime, timedelta +from typing import Any + +import httpx +import pytest +from fastapi import FastAPI + +from kit.health import Registry +from kit.httpapi import install_conventions +from kit.messaging import ( + EVENT_ID, + OCCURRED_AT, + TENANT, + Broker, + BrokerTimeout, + BrokerUnavailable, + Envelope, + MessageBus, + envelope, + register_brokers, +) +from kit.messaging._broker import Subscription, _delivery_count +from kit.observability import _metrics + +WHEN = datetime(2026, 9, 7, 12, 0, tzinfo=UTC) + + +def an_envelope(**overrides: Any) -> Envelope: + fields: dict[str, Any] = { + "tenant": "org_1", + "event_id": "evt_1", + "event_type": "repository.indexed", + "occurred_at": WHEN, + } + fields.update(overrides) + return Envelope(**fields) + + +class FakeMetadata: + def __init__(self, delivered: int) -> None: + self.num_delivered = delivered + + +class FakeMessage: + """One delivery, recording how it was settled.""" + + def __init__( + self, + headers: dict[str, str] | None = None, + data: bytes = b"{}", + delivered: int = 1, + fails: bool = False, + ) -> None: + self.headers = headers if headers is not None else an_envelope().headers() + self.data = data + self.metadata = FakeMetadata(delivered) + self.settled: list[tuple[str, float | None]] = [] + self._fails = fails + + async def ack(self) -> None: + self._record("ack") + + async def nak(self, delay: float | None = None) -> None: + self._record("nak", delay) + + async def term(self) -> None: + self._record("term") + + def _record(self, outcome: str, delay: float | None = None) -> None: + if self._fails: + raise RuntimeError("the connection went away") + self.settled.append((outcome, delay)) + + +class FakeSubscription: + def __init__(self, batches: list[Any]) -> None: + self._batches = batches + + async def fetch(self, batch: int, timeout: float) -> Any: # noqa: ASYNC109 + result = self._batches.pop(0) + if isinstance(result, Exception): + raise result + return result + + +class FakeStream: + def __init__(self, *, publish_error: Exception | None = None) -> None: + self.published: list[tuple[str, bytes, dict[str, str], float]] = [] + self._publish_error = publish_error + self.subscription = FakeSubscription([]) + self.subscribe_error: Exception | None = None + + async def publish( + self, + subject: str, + payload: bytes, + headers: dict[str, str], + timeout: float, # noqa: ASYNC109 + ) -> None: + if self._publish_error is not None: + raise self._publish_error + self.published.append((subject, payload, headers, timeout)) + + async def pull_subscribe(self, subject: str, durable: str, stream: str) -> Any: + if self.subscribe_error is not None: + raise self.subscribe_error + return self.subscription + + +class FakeConnection: + def __init__(self, stream: FakeStream | None = None, *, drain_fails: bool = False) -> None: + self.is_connected = True + self._stream = stream or FakeStream() + self.drained = False + self._drain_fails = drain_fails + + def jetstream(self) -> FakeStream: + return self._stream + + async def drain(self) -> None: + if self._drain_fails: + raise RuntimeError("drain refused") + self.drained = True + + +def a_bus(connection: FakeConnection | None = None) -> MessageBus: + return MessageBus(Broker(name="events", servers=("nats://x:4222",)), connection) + + +# --- the envelope ------------------------------------------------------------ + + +def test_the_four_headers_go_out_and_come_back() -> None: + """A round trip, because the wire form is the contract between services.""" + read = envelope(an_envelope().headers()) + + assert read == an_envelope() + + +def test_the_event_id_rides_as_the_header_jetstream_dedupes_on() -> None: + """NOT one of our own names, and that is the whole point: the server refuses + a second publish carrying an id it has already seen.""" + assert EVENT_ID == "Nats-Msg-Id" + assert an_envelope().headers()[EVENT_ID] == "evt_1" + + +def test_an_occurred_at_is_normalised_to_utc() -> None: + """Two publishers in two zones must produce the same string for one moment.""" + elsewhere = WHEN.astimezone(_offset(hours=5)) + + assert an_envelope(occurred_at=elsewhere).headers()[OCCURRED_AT] == WHEN.isoformat() + + +@pytest.mark.parametrize("blank", ["", " "]) +@pytest.mark.parametrize("field", ["tenant", "event_id", "event_type"]) +def test_a_blank_required_field_is_refused(field: str, blank: str) -> None: + """A blank tenant scopes to nothing and a blank id defeats deduplication. + Both are publisher bugs that would otherwise be found by the consumer.""" + with pytest.raises(ValueError, match=field): + an_envelope(**{field: blank}) + + +def test_a_naive_timestamp_is_refused_rather_than_assumed_utc() -> None: + """Reading it as UTC would be a guess that then looks like data.""" + with pytest.raises(ValueError, match="timezone aware"): + an_envelope(occurred_at=datetime(2026, 9, 7, 12, 0)) # noqa: DTZ001 + + +def test_an_unparsable_timestamp_names_the_header() -> None: + with pytest.raises(ValueError, match=OCCURRED_AT): + envelope({**an_envelope().headers(), OCCURRED_AT: "yesterday"}) + + +def test_absent_headers_are_refused_by_name() -> None: + """`None` is what a message with no headers at all carries.""" + with pytest.raises(ValueError): + envelope(None) + + +# --- publishing -------------------------------------------------------------- + + +async def test_publishing_sends_the_headers_and_waits_for_the_stream() -> None: + """Waiting is the point: an unwaited publish reports success while the + message is still in flight.""" + connection = FakeConnection() + bus = a_bus(connection) + + await bus.publish("events.repository.indexed", b"payload", an_envelope()) + + subject, payload, headers, timeout = connection.jetstream().published[0] + assert subject == "events.repository.indexed" + assert payload == b"payload" + assert headers[TENANT] == "org_1" + assert timeout == bus.broker.publish_timeout_seconds + + +async def test_a_publish_timeout_is_its_own_type() -> None: + """Separate from Unavailable because it means we stopped, not that the bus + said no, and the two lead different places.""" + bus = a_bus(FakeConnection(FakeStream(publish_error=TimeoutError()))) + + with pytest.raises(BrokerTimeout, match="events"): + await bus.publish("events.x", b"", an_envelope()) + + +async def test_a_failed_publish_names_the_broker() -> None: + bus = a_bus(FakeConnection(FakeStream(publish_error=RuntimeError("no stream")))) + + with pytest.raises(BrokerUnavailable, match="events"): + await bus.publish("events.x", b"", an_envelope()) + + +async def test_publishing_before_connecting_is_refused() -> None: + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + with pytest.raises(BrokerUnavailable, match="not connected"): + await bus.publish("events.x", b"", an_envelope()) + + +# --- consuming --------------------------------------------------------------- + + +async def test_fetch_returns_unsettled_messages() -> None: + """Nothing is acknowledged on the caller's behalf, so a handler that raises + gets its message back rather than losing it.""" + raw = FakeMessage() + subscription = Subscription("events", "EVENTS", FakeSubscription([[raw]])) + + [message] = await subscription.fetch() + + assert message.envelope.tenant == "org_1" + assert message.payload == b"{}" + assert raw.settled == [] + + +async def test_an_empty_batch_is_not_an_error() -> None: + """The ordinary case on a quiet stream. The client signals it as a timeout.""" + subscription = Subscription("events", "EVENTS", FakeSubscription([TimeoutError()])) + + assert await subscription.fetch() == () + + +async def test_a_failed_fetch_names_the_broker() -> None: + subscription = Subscription("events", "EVENTS", FakeSubscription([RuntimeError("gone")])) + + with pytest.raises(BrokerUnavailable, match="events"): + await subscription.fetch() + + +async def test_an_unreadable_message_is_terminated_not_naked() -> None: + """It will be malformed next time too, so a nak would spend the whole + attempt budget rediscovering that.""" + bad = FakeMessage(headers={TENANT: "org_1"}) + subscription = Subscription("events", "EVENTS", FakeSubscription([[bad]])) + + assert await subscription.fetch() == () + assert bad.settled == [("term", None)] + + +async def test_terminating_an_unreadable_message_never_raises() -> None: + """The connection may already be gone; that must not become the caller's + problem on a message we were dropping anyway.""" + bad = FakeMessage(headers={}, fails=True) + subscription = Subscription("events", "EVENTS", FakeSubscription([[bad]])) + + assert await subscription.fetch() == () + + +async def test_a_readable_message_survives_beside_an_unreadable_one() -> None: + """One poisonous message must not cost the batch.""" + subscription = Subscription( + "events", "EVENTS", FakeSubscription([[FakeMessage(headers={}), FakeMessage()]]) + ) + + assert len(await subscription.fetch()) == 1 + + +# --- settling ---------------------------------------------------------------- + + +async def test_the_three_settle_verbs_reach_the_message() -> None: + raw = [FakeMessage(), FakeMessage(), FakeMessage()] + subscription = Subscription("events", "EVENTS", FakeSubscription([raw])) + one, two, three = await subscription.fetch(batch=3) + + await one.ack() + await two.nak(delay_seconds=30.0) + await three.term() + + assert raw[0].settled == [("ack", None)] + assert raw[1].settled == [("nak", 30.0)] + assert raw[2].settled == [("term", None)] + + +async def test_a_failed_settle_does_not_raise() -> None: + """The work happened; only the acknowledgement did not. Raising would tell a + caller its work failed, which is worse than the truth: the message comes back + and the handler's idempotency covers it.""" + subscription = Subscription("events", "EVENTS", FakeSubscription([[FakeMessage(fails=True)]])) + [message] = await subscription.fetch() + + await message.ack() + + +async def test_a_redelivery_reports_which_attempt_it_is() -> None: + """Above one is how a handler tells poisonous from merely slow.""" + subscription = Subscription("events", "EVENTS", FakeSubscription([[FakeMessage(delivered=4)]])) + + [message] = await subscription.fetch() + + assert message.delivered == 4 + + +@pytest.mark.parametrize("metadata", [None, FakeMetadata(0), "not metadata"]) +def test_absent_delivery_metadata_reads_as_a_first_delivery(metadata: Any) -> None: + """Zero would say "never delivered" about a message being held.""" + + class Bare: + pass + + message = Bare() + if metadata is not None: + message.metadata = metadata # type: ignore[attr-defined] + + assert _delivery_count(message) == 1 + + +# --- connecting -------------------------------------------------------------- + + +async def test_connecting_twice_does_not_open_a_second_connection() -> None: + connection = FakeConnection() + bus = a_bus(connection) + + await bus.connect() + + assert bus.connected + + +async def test_connecting_with_no_servers_is_refused() -> None: + bus = MessageBus(Broker(name="events")) + + with pytest.raises(BrokerUnavailable, match="no servers"): + await bus.connect() + + +async def test_connecting_without_the_extra_names_the_extra( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """The failure a service that forgot the extra can actually read.""" + real = builtins.__import__ + + def missing(name: str, *args: Any, **kwargs: Any) -> Any: + if name == "nats": + raise ImportError(name) + return real(name, *args, **kwargs) + + monkeypatch.setattr(builtins, "__import__", missing) + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + with pytest.raises(BrokerUnavailable, match=r"scadable-kit\[nats\]"): + await bus.connect() + + +class FakeNats: + """Stands in for the `nats` module, so the real connect path is exercised.""" + + def __init__(self, connection: Any = None, error: Exception | None = None) -> None: + self.connection = connection + self.error = error + self.options: dict[str, Any] = {} + + async def connect(self, **options: Any) -> Any: + self.options = options + if self.error is not None: + raise self.error + return self.connection + + +@pytest.fixture +def nats_module(monkeypatch: pytest.MonkeyPatch): + """Install a stub `nats` module for the duration of one test.""" + + def install(fake: FakeNats) -> FakeNats: + module = types.ModuleType("nats") + module.connect = fake.connect # type: ignore[attr-defined] + monkeypatch.setitem(sys.modules, "nats", module) + return fake + + return install + + +async def test_connecting_passes_the_reconnect_settings_through(nats_module: Any) -> None: + """Retrying forever is deliberate: a consumer that gives up becomes a + process that is running, healthy by any liveness probe, and consuming + nothing.""" + fake = nats_module(FakeNats(FakeConnection())) + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + await bus.connect() + + assert fake.options["max_reconnect_attempts"] == -1 + assert fake.options["servers"] == ["nats://x:4222"] + assert "user_credentials" not in fake.options + assert "token" not in fake.options + assert bus.connected + + +async def test_credentials_are_passed_when_set(nats_module: Any) -> None: + """Carried even though the deployed bus has no auth, so turning it on is a + config change in one service rather than a release of this package.""" + fake = nats_module(FakeNats(FakeConnection())) + bus = MessageBus( + Broker( + name="events", + servers=("nats://x:4222",), + credentials_path="/etc/nats/creds", + token="t", # noqa: S106 + ) + ) + + await bus.connect() + + assert fake.options["user_credentials"] == "/etc/nats/creds" + assert fake.options["token"] == "t" # noqa: S105 + + +async def test_a_connect_timeout_is_its_own_type(nats_module: Any) -> None: + nats_module(FakeNats(error=TimeoutError())) + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + with pytest.raises(BrokerTimeout, match="connect timed out"): + await bus.connect() + + +async def test_a_refused_connect_names_the_broker(nats_module: Any) -> None: + nats_module(FakeNats(error=RuntimeError("no route to host"))) + bus = MessageBus(Broker(name="events", servers=("nats://x:4222",))) + + with pytest.raises(BrokerUnavailable, match="events"): + await bus.connect() + + +async def test_connecting_an_already_connected_bus_is_a_no_op(nats_module: Any) -> None: + """A composition root that retries should not leak a second connection.""" + fake = nats_module(FakeNats(FakeConnection())) + bus = a_bus(FakeConnection()) + + await bus.connect() + + assert fake.options == {} + + +async def test_a_broker_needs_a_name() -> None: + """It labels metrics and health checks, so a blank one is a series nobody + can find.""" + with pytest.raises(ValueError, match="name"): + Broker(name=" ") + + +def test_configured_answers_without_opening_a_socket() -> None: + assert not Broker(name="events").configured + assert Broker(name="events", servers=("nats://x:4222",)).configured + + +# --- subscribing ------------------------------------------------------------- + + +async def test_subscribing_binds_an_existing_durable_consumer() -> None: + """Binding rather than creating keeps ack policy and max deliveries in one + place instead of in whichever service connected first.""" + bus = a_bus(FakeConnection()) + + subscription = await bus.subscribe(stream="EVENTS", subject="events.>", durable="brain") + + assert subscription.stream == "EVENTS" + + +async def test_binding_a_missing_consumer_names_it() -> None: + stream = FakeStream() + stream.subscribe_error = RuntimeError("consumer not found") + bus = a_bus(FakeConnection(stream)) + + with pytest.raises(BrokerUnavailable, match="brain"): + await bus.subscribe(stream="EVENTS", subject="events.>", durable="brain") + + +# --- closing ----------------------------------------------------------------- + + +async def test_closing_drains_so_in_flight_work_finishes() -> None: + """A bare close abandons held messages and they come back as redeliveries.""" + connection = FakeConnection() + bus = a_bus(connection) + + await bus.close() + + assert connection.drained + assert not bus.connected + + +async def test_closing_an_unconnected_bus_is_a_no_op() -> None: + await MessageBus(Broker(name="events")).close() + + +async def test_a_failed_drain_does_not_raise_on_the_way_down() -> None: + bus = a_bus(FakeConnection(drain_fails=True)) + + await bus.close() + + assert not bus.connected + + +# --- readiness --------------------------------------------------------------- + + +async def readiness_of(registry: Registry) -> httpx.Response: + app = FastAPI() + install_conventions(app, readiness=registry) + transport = httpx.ASGITransport(app=app) + async with httpx.AsyncClient(transport=transport, base_url="http://test") as client: + return await client.get("/readyz") + + +async def test_a_connected_bus_reports_ready() -> None: + registry = Registry() + register_brokers(registry, [a_bus(FakeConnection())]) + + response = await readiness_of(registry) + + assert response.json()["checks"]["events"] == "ready" + + +async def test_a_disconnected_bus_is_visible_but_does_not_block() -> None: + """THE decision, asserted. A replica that cannot reach the bus still serves + every read it owns, so failing readiness would turn one component's outage + into two and would not reconnect anything.""" + registry = Registry() + register_brokers(registry, [MessageBus(Broker(name="events"))]) + + response = await readiness_of(registry) + + assert response.status_code == 200, "a disconnected bus took the service out of rotation" + assert response.json()["status"] == "ready" + assert response.json()["checks"]["events"] == "error" + + +# --- metrics ----------------------------------------------------------------- + + +class Recorder: + def __init__(self) -> None: + self.calls: list[tuple[float, dict[str, Any]]] = [] + + def add(self, value: float, attributes: dict[str, Any]) -> None: + self.calls.append((value, attributes)) + + +@pytest.fixture +def recorders(monkeypatch: pytest.MonkeyPatch) -> dict[str, Recorder]: + """Every messaging instrument, so a name absent from `_build` shows up here + as a missing key rather than as silence.""" + made = { + name: Recorder() + for name in ( + _metrics.MESSAGING_PUBLISHED, + _metrics.MESSAGING_CONSUMED, + _metrics.MESSAGING_SETTLED, + _metrics.MESSAGING_DELIVERIES, + ) + } + monkeypatch.setattr(_metrics, "_instruments", dict(made)) + return made + + +async def test_the_metric_label_is_the_stream_never_the_subject( + recorders: dict[str, Recorder], +) -> None: + """A subject carries a tenant id in most useful designs, and one label + holding one turns a single series into millions.""" + await a_bus(FakeConnection()).publish("events.repository.indexed", b"", an_envelope()) + + _, attributes = recorders[_metrics.MESSAGING_PUBLISHED].calls[0] + assert attributes == {"broker": "events", "stream": "events"} + + +async def test_settling_is_counted_by_outcome(recorders: dict[str, Recorder]) -> None: + subscription = Subscription("events", "EVENTS", FakeSubscription([[FakeMessage()]])) + [message] = await subscription.fetch() + + await message.nak() + + _, attributes = recorders[_metrics.MESSAGING_SETTLED].calls[0] + assert attributes["outcome"] == "nak" + + +async def test_a_failed_settle_is_not_counted_as_settled( + recorders: dict[str, Recorder], +) -> None: + """Counting it would report an acknowledgement that never reached the bus.""" + subscription = Subscription("events", "EVENTS", FakeSubscription([[FakeMessage(fails=True)]])) + [message] = await subscription.fetch() + + await message.ack() + + assert recorders[_metrics.MESSAGING_SETTLED].calls == [] + + +def _offset(hours: int) -> Any: + """A fixed offset, without importing zoneinfo for one assertion.""" + return __import__("datetime").timezone(timedelta(hours=hours)) diff --git a/uv.lock b/uv.lock index 9c3fa0e..c8ac7ec 100644 --- a/uv.lock +++ b/uv.lock @@ -243,6 +243,15 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/cb/b1/3846dd7f199d53cb17f49cba7e651e9ce294d8497c8c150530ed11865bb8/iniconfig-2.3.0-py3-none-any.whl", hash = "sha256:f631c04d2c48c52b84d0d0549c99ff3859c98df65b3101406327ecc7d53fbf12", size = 7484, upload-time = "2025-10-18T21:55:41.639Z" }, ] +[[package]] +name = "nats-py" +version = "2.15.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/02/f0/fc5e93f2b0dd14a202590ad9d30eda1955ea872039b5204357348d0f4b1e/nats_py-2.15.0.tar.gz", hash = "sha256:6622c547d9a7d2313d9c147d46c386188f4ec2c7b5c9f9a0438a4d1b55f54a93", size = 75995, upload-time = "2026-06-05T07:34:03.904Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/db/a8/b55606c7c621fb813c8ec78baf201d2c78bf6051091ec0c7ada572999e95/nats_py-2.15.0-py3-none-any.whl", hash = "sha256:9f8d36aa52a9926a88b8f1d70cf1fdce0ad387941479b500ee9ab3e51073cefd", size = 90334, upload-time = "2026-06-05T07:34:02.81Z" }, +] + [[package]] name = "nodeenv" version = "1.10.0" @@ -623,7 +632,7 @@ wheels = [ [[package]] name = "scadable-kit" -version = "0.6.1" +version = "0.7.0" source = { editable = "." } dependencies = [ { name = "fastapi" }, @@ -633,6 +642,9 @@ dependencies = [ ] [package.optional-dependencies] +nats = [ + { name = "nats-py" }, +] otel = [ { name = "opentelemetry-exporter-otlp-proto-http" }, { name = "opentelemetry-instrumentation-fastapi" }, @@ -654,6 +666,7 @@ dev = [ requires-dist = [ { name = "fastapi", specifier = ">=0.141,<0.142" }, { name = "httpx", specifier = ">=0.28,<0.29" }, + { name = "nats-py", marker = "extra == 'nats'", specifier = ">=2.15,<3" }, { name = "opentelemetry-exporter-otlp-proto-http", marker = "extra == 'otel'", specifier = ">=1.44,<2" }, { name = "opentelemetry-instrumentation-fastapi", marker = "extra == 'otel'", specifier = ">=0.65b0" }, { name = "opentelemetry-instrumentation-httpx", marker = "extra == 'otel'", specifier = ">=0.65b0" }, @@ -661,7 +674,7 @@ requires-dist = [ { name = "pydantic", specifier = ">=2.13,<3" }, { name = "pydantic-settings", specifier = ">=2.7,<3" }, ] -provides-extras = ["otel"] +provides-extras = ["otel", "nats"] [package.metadata.requires-dev] dev = [