# Copyright 2021 The NATS Authors # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. # from __future__ import annotations import datetime from dataclasses import dataclass, fields, replace from enum import Enum from typing import Any, Dict, Iterable, Iterator, List, Optional, TypeVar _NANOSECOND = 10**9 class Header(str, Enum): CONSUMER_STALLED = "Nats-Consumer-Stalled" DESCRIPTION = "Description" EXPECTED_LAST_MSG_ID = "Nats-Expected-Last-Msg-Id" EXPECTED_LAST_SEQUENCE = "Nats-Expected-Last-Sequence" EXPECTED_LAST_SUBJECT_SEQUENCE = "Nats-Expected-Last-Subject-Sequence" EXPECTED_STREAM = "Nats-Expected-Stream" LAST_CONSUMER = "Nats-Last-Consumer" LAST_STREAM = "Nats-Last-Stream" MSG_ID = "Nats-Msg-Id" MSG_TTL = "Nats-TTL" ROLLUP = "Nats-Rollup" STATUS = "Status" DEFAULT_PREFIX = "$JS.API" INBOX_PREFIX = b"_INBOX." class StatusCode(str, Enum): SERVICE_UNAVAILABLE = "503" NO_MESSAGES = "404" REQUEST_TIMEOUT = "408" CONFLICT = "409" CONTROL_MESSAGE = "100" _B = TypeVar("_B", bound="Base") @dataclass class Base: """ Helper dataclass to filter unknown fields from the API. """ @staticmethod def _convert(resp: Dict[str, Any], field: str, type: type[Base]) -> None: """Convert the field into the given type in place.""" data = resp.get(field, None) if data is None: resp[field] = None elif isinstance(data, list): resp[field] = [type.from_response(item) for item in data] else: resp[field] = type.from_response(data) @staticmethod def _convert_nanoseconds(resp: Dict[str, Any], field: str) -> None: """Convert the given field from nanoseconds to seconds in place.""" val = resp.get(field, None) if val is not None: val = val / _NANOSECOND resp[field] = val @staticmethod def _to_nanoseconds(val: Optional[float]) -> Optional[int]: """Convert the value from seconds to nanoseconds.""" if val is None: # We use 0 to avoid sending null to Go servers. return 0 return int(val * _NANOSECOND) @staticmethod def _convert_utc_iso(resp: Dict[str, Any], field: str) -> None: """Convert a ISO 8601/RFC 3339 string field into a datetime in place.""" val = resp.get(field, None) if val is not None: resp[field] = Base._parse_utc_iso(val) @staticmethod def _to_utc_iso(date: datetime.datetime) -> str: """Convert a datetime into an ISO 8601/RFC 3339 formatted string. If datetime does not have timezone information, datetime is assumed to be in UTC timezone. Strings are passed through as-is for backwards compatibility. """ if isinstance(date, str): return date if date.tzinfo is None: date = date.replace(tzinfo=datetime.timezone.utc) elif date.tzinfo != datetime.timezone.utc: date = date.astimezone(datetime.timezone.utc) return date.isoformat().replace("+00:00", "Z").replace(".000000", "") @staticmethod def _parse_utc_iso(time_string: str) -> datetime.datetime: """Parse an ISO 8601 timestamp (with nanoseconds) into a UTC datetime.""" # Replace Z with UTC offset s = time_string.replace("Z", "+00:00") # Trim fractional seconds to 6 digits (microsecond precision) date_part, frac_tz = s.split(".", 1) frac, tz = frac_tz.split("+") frac = frac[:6] # keep only microseconds s = f"{date_part}.{frac}+{tz}" return datetime.datetime.fromisoformat(s).astimezone(datetime.timezone.utc) @classmethod def from_response(cls: type[_B], resp: Dict[str, Any]) -> _B: """Read the class instance from a server response. Unknown fields are ignored ("open-world assumption"). """ params = {} for field in fields(cls): if field.name in resp: params[field.name] = resp[field.name] return cls(**params) def evolve(self: _B, **params) -> _B: """Return a copy of the instance with the passed values replaced.""" return replace(self, **params) def as_dict(self) -> Dict[str, object]: """Return the object converted into an API-friendly dict.""" result = {} for field in fields(self): val = getattr(self, field.name) if val is None: continue if isinstance(val, Base): val = val.as_dict() if isinstance(val, list): if len(val) > 0 and isinstance(val[0], Base): val = [v.as_dict() for v in val if isinstance(v, Base)] result[field.name] = val return result @dataclass class PubAck(Base): """ PubAck is the response of publishing a message to JetStream. """ stream: str seq: int domain: Optional[str] = None duplicate: Optional[bool] = None @dataclass class Placement(Base): """Placement directives to consider when placing replicas of this stream""" cluster: Optional[str] = None tags: Optional[List[str]] = None @dataclass class ExternalStream(Base): api: str deliver: Optional[str] = None def as_dict(self) -> Dict[str, object]: result = super().as_dict() return result @dataclass class StreamSource(Base): name: str opt_start_seq: Optional[int] = None opt_start_time: Optional[datetime.datetime] = None filter_subject: Optional[str] = None external: Optional[ExternalStream] = None subject_transforms: Optional[List[SubjectTransform]] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "external", ExternalStream) cls._convert(resp, "subject_transforms", SubjectTransform) cls._convert_utc_iso(resp, "opt_start_time") return super().from_response(resp) def as_dict(self) -> Dict[str, object]: result = super().as_dict() if self.subject_transforms: result["subject_transforms"] = [tr.as_dict() for tr in self.subject_transforms] if self.opt_start_time is not None: result["opt_start_time"] = self._to_utc_iso(self.opt_start_time) return result @dataclass class StreamSourceInfo(Base): name: str lag: Optional[int] = None active: Optional[int] = None error: Optional[Dict[str, Any]] = None @dataclass class LostStreamData(Base): msgs: Optional[List[int]] = None bytes: Optional[int] = None @dataclass class StreamState(Base): messages: int bytes: int first_seq: int last_seq: int consumer_count: int deleted: Optional[List[int]] = None num_deleted: Optional[int] = None lost: Optional[LostStreamData] = None subjects: Optional[Dict[str, int]] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "lost", LostStreamData) return super().from_response(resp) class RetentionPolicy(str, Enum): """How message retention is considered""" LIMITS = "limits" INTEREST = "interest" WORK_QUEUE = "workqueue" class StorageType(str, Enum): """The type of storage backend""" FILE = "file" MEMORY = "memory" class DiscardPolicy(str, Enum): """Discard policy when a stream reaches its limits""" OLD = "old" NEW = "new" class StoreCompression(str, Enum): """ If stream is file-based and a compression algorithm is specified, the stream data will be compressed on disk. Valid options are none or s2 for Snappy compression. Introduced in nats-server 2.10.0. """ NONE = "none" S2 = "s2" class PersistMode(str, Enum): """ PersistMode defines the consistency and durability guarantees for stream persistence. See ADR-56 for details: https://github.com/nats-io/nats-architecture-and-design/blob/main/adr/ADR-56.md Currently only applicable to R1 (single replica) streams. Introduced in nats-server 2.12.0. """ # DEFAULT represents the strongest consistency guarantee. # Uses synchronous writes with fsync for maximum durability. # Server does not store this value - it's the implied default when unset. DEFAULT = "default" # ASYNC enables asynchronous flushing of data to disk. # Returns PubAck before disk persistence occurs, batching writes in memory. # Provides significantly improved performance at the cost of potential data loss # during infrastructure failures. Incompatible with batch publishing. ASYNC = "async" @dataclass class RePublish(Base): """ RePublish is for republishing messages once committed to a stream. The original subject cis remapped from the subject pattern to the destination pattern. """ src: Optional[str] = None dest: Optional[str] = None headers_only: Optional[bool] = None @dataclass class SubjectTransform(Base): """Subject transform to apply to matching messages.""" src: str dest: str def as_dict(self) -> Dict[str, object]: result = super().as_dict() return result @dataclass class StreamConfig(Base): """ StreamConfig represents the configuration of a stream. """ name: Optional[str] = None description: Optional[str] = None subjects: Optional[List[str]] = None retention: Optional[RetentionPolicy] = None max_consumers: Optional[int] = None max_msgs: Optional[int] = None max_bytes: Optional[int] = None discard: Optional[DiscardPolicy] = DiscardPolicy.OLD discard_new_per_subject: bool = False max_age: Optional[float] = None # in seconds max_msgs_per_subject: int = -1 max_msg_size: Optional[int] = -1 storage: Optional[StorageType] = None num_replicas: Optional[int] = None no_ack: bool = False template_owner: Optional[str] = None duplicate_window: float = 0 placement: Optional[Placement] = None mirror: Optional[StreamSource] = None sources: Optional[List[StreamSource]] = None sealed: bool = False deny_delete: bool = False deny_purge: bool = False allow_rollup_hdrs: bool = False # Allow republish of the message after being sequenced and stored. republish: Optional[RePublish] = None subject_transform: Optional[SubjectTransform] = None # Allow higher performance, direct access to get individual messages. E.g. KeyValue allow_direct: Optional[bool] = None # Allow higher performance and unified direct access for mirrors as well. mirror_direct: Optional[bool] = None # Allow compressing messages. compression: Optional[StoreCompression] = None # Allow per-message TTL via Nats-TTL header. Introduced in nats-server 2.11.0. allow_msg_ttl: Optional[bool] = None # Allow scheduled/delayed messages. Introduced in nats-server 2.12.0. allow_msg_schedules: Optional[bool] = None # Allow atomic batch publishing. Introduced in nats-server 2.12.0. allow_atomic: Optional[bool] = None # Allow batched publishing. Introduced in nats-server 2.12.0. allow_batched: Optional[bool] = None # Persistence mode for stream. Only applicable to R1 streams. # Introduced in nats-server 2.12.0. persist_mode: Optional[PersistMode] = None # Metadata are user defined string key/value pairs. metadata: Optional[Dict[str, str]] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert_nanoseconds(resp, "max_age") cls._convert_nanoseconds(resp, "duplicate_window") cls._convert(resp, "placement", Placement) cls._convert(resp, "mirror", StreamSource) cls._convert(resp, "sources", StreamSource) cls._convert(resp, "republish", RePublish) cls._convert(resp, "subject_transform", SubjectTransform) return super().from_response(resp) def as_dict(self) -> Dict[str, object]: result = super().as_dict() result["duplicate_window"] = self._to_nanoseconds(self.duplicate_window) result["max_age"] = self._to_nanoseconds(self.max_age) if self.sources: result["sources"] = [src.as_dict() for src in self.sources] if self.compression and (self.compression != StoreCompression.NONE and self.compression != StoreCompression.S2): raise ValueError("nats: invalid store compression type: %s" % self.compression) if self.metadata and not isinstance(self.metadata, dict): raise ValueError("nats: invalid metadata format") return result @dataclass class PeerInfo(Base): name: Optional[str] = None current: Optional[bool] = None offline: Optional[bool] = None active: Optional[int] = None lag: Optional[int] = None @dataclass class ClusterInfo(Base): leader: Optional[str] = None name: Optional[str] = None replicas: Optional[List[PeerInfo]] = None raft_group: Optional[str] = None leader_since: Optional[datetime.datetime] = None traffic_acc: Optional[str] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "replicas", PeerInfo) if "leader_since" in resp and resp["leader_since"]: resp["leader_since"] = cls._parse_utc_iso(resp["leader_since"]) return super().from_response(resp) @dataclass class StreamInfo(Base): """ StreamInfo is the latest information about a stream from JetStream. """ config: StreamConfig state: StreamState mirror: Optional[StreamSourceInfo] = None sources: Optional[List[StreamSourceInfo]] = None cluster: Optional[ClusterInfo] = None did_create: Optional[bool] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "config", StreamConfig) cls._convert(resp, "state", StreamState) cls._convert(resp, "mirror", StreamSourceInfo) cls._convert(resp, "sources", StreamSourceInfo) cls._convert(resp, "cluster", ClusterInfo) return super().from_response(resp) @dataclass class StreamsListIterator(Iterable): """ StreamsListIterator is an iterator for streams list responses from JetStream. """ def __init__(self, offset: int, total: int, streams: List[Dict[str, any]]) -> None: self.offset = offset self.total = total self.streams = streams self._index = 0 def __iter__(self) -> Iterator[StreamInfo]: return self def __next__(self) -> StreamInfo: if self._index < len(self.streams): stream_info = StreamInfo.from_response(self.streams[self._index]) self._index += 1 return stream_info else: raise StopIteration class AckPolicy(str, Enum): """Policies defining how messages should be acknowledged. If an ack is required but is not received within the AckWait window, the message will be redelivered. References: * `Consumers, AckPolicy <https://docs.nats.io/jetstream/concepts/consumers#ackpolicy>`_ """ NONE = "none" ALL = "all" EXPLICIT = "explicit" class DeliverPolicy(str, Enum): """When a consumer is first created, it can specify where in the stream it wants to start receiving messages. This is the DeliverPolicy, and this enumeration defines allowed values. References: * `Consumers, DeliverPolicy/OptStartSeq/OptStartTime <https://docs.nats.io/jetstream/concepts/consumers#deliverpolicy-optstartseq-optstarttime>`_ """ # noqa: E501 ALL = "all" LAST = "last" NEW = "new" BY_START_SEQUENCE = "by_start_sequence" BY_START_TIME = "by_start_time" LAST_PER_SUBJECT = "last_per_subject" class ReplayPolicy(str, Enum): """The replay policy applies when the DeliverPolicy is one of: * all * by_start_sequence * by_start_time since those deliver policies begin reading the stream at a position other than the end. References: * `Consumers, ReplayPolicy <https://docs.nats.io/jetstream/concepts/consumers#replaypolicy>`_ """ INSTANT = "instant" ORIGINAL = "original" @dataclass class ConsumerConfig(Base): """Consumer configuration. References: * `Consumers <https://docs.nats.io/jetstream/concepts/consumers>`_ """ name: Optional[str] = None durable_name: Optional[str] = None description: Optional[str] = None deliver_policy: Optional[DeliverPolicy] = DeliverPolicy.ALL opt_start_seq: Optional[int] = None opt_start_time: Optional[datetime.datetime] = None ack_policy: Optional[AckPolicy] = AckPolicy.EXPLICIT ack_wait: Optional[float] = None # in seconds max_deliver: Optional[int] = None backoff: Optional[List[float]] = None # in seconds, overrides ack_wait filter_subject: Optional[str] = None filter_subjects: Optional[List[str]] = None replay_policy: Optional[ReplayPolicy] = ReplayPolicy.INSTANT rate_limit_bps: Optional[int] = None sample_freq: Optional[str] = None max_waiting: Optional[int] = None max_ack_pending: Optional[int] = None flow_control: Optional[bool] = None idle_heartbeat: Optional[float] = None headers_only: Optional[bool] = None # Push based consumers. deliver_subject: Optional[str] = None # Push based queue consumers. deliver_group: Optional[str] = None # Ephemeral inactivity threshold inactive_threshold: Optional[float] = None # in seconds # Generally inherited by parent stream and other markers, now can # be configured directly. num_replicas: Optional[int] = None # Force memory storage. mem_storage: Optional[bool] = None # Metadata are user defined string key/value pairs. metadata: Optional[Dict[str, str]] = None # Consumer pause until timestamp. # Temporarily suspend message delivery until the specified time (RFC 3339 format). # Introduced in nats-server 2.11.0. pause_until: Optional[str] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert_nanoseconds(resp, "ack_wait") cls._convert_nanoseconds(resp, "idle_heartbeat") cls._convert_nanoseconds(resp, "inactive_threshold") cls._convert_utc_iso(resp, "opt_start_time") if "backoff" in resp: resp["backoff"] = [val / _NANOSECOND for val in resp["backoff"]] return super().from_response(resp) def as_dict(self) -> Dict[str, object]: result = super().as_dict() if self.opt_start_time is not None: result["opt_start_time"] = self._to_utc_iso(self.opt_start_time) result["ack_wait"] = self._to_nanoseconds(self.ack_wait) result["idle_heartbeat"] = self._to_nanoseconds(self.idle_heartbeat) result["inactive_threshold"] = self._to_nanoseconds(self.inactive_threshold) if self.backoff: result["backoff"] = [self._to_nanoseconds(i) for i in self.backoff] return result @dataclass class SequenceInfo(Base): consumer_seq: int stream_seq: int last_active: Optional[datetime.datetime] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert_utc_iso(resp, "last_active") return super().from_response(resp) def as_dict(self) -> Dict[str, object]: result = super().as_dict() if self.last_active is not None: result["last_active"] = self._to_utc_iso(self.last_active) return result @dataclass class ConsumerInfo(Base): """ ConsumerInfo represents the info about the consumer. """ name: str stream_name: str config: ConsumerConfig created: datetime.datetime delivered: Optional[SequenceInfo] = None ack_floor: Optional[SequenceInfo] = None num_ack_pending: Optional[int] = None num_redelivered: Optional[int] = None num_waiting: Optional[int] = None num_pending: Optional[int] = None cluster: Optional[ClusterInfo] = None push_bound: Optional[bool] = None # Indicates if the consumer is currently paused. # Introduced in nats-server 2.11.0. paused: Optional[bool] = None # RFC 3339 timestamp until which the consumer is paused. # Introduced in nats-server 2.11.0. pause_remaining: Optional[str] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "delivered", SequenceInfo) cls._convert(resp, "ack_floor", SequenceInfo) cls._convert(resp, "config", ConsumerConfig) cls._convert(resp, "cluster", ClusterInfo) cls._convert_utc_iso(resp, "created") return super().from_response(resp) def as_dict(self) -> Dict[str, object]: result = super().as_dict() result["created"] = self._to_utc_iso(self.created) return result @dataclass class ConsumerPause(Base): """ ConsumerPause represents the pause state after a pause or resume operation. Introduced in nats-server 2.11.0. """ paused: bool pause_until: Optional[str] = None pause_remaining: Optional[str] = None @dataclass class AccountLimits(Base): """Account limits References: * `Multi-tenancy & Resource Mgmt <https://docs.nats.io/jetstream/resource_management>`_ """ max_memory: int max_storage: int max_streams: int max_consumers: int max_ack_pending: int memory_max_stream_bytes: int storage_max_stream_bytes: int max_bytes_required: bool @dataclass class Tier(Base): memory: int storage: int streams: int consumers: int limits: AccountLimits @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "limits", AccountLimits) return super().from_response(resp) @dataclass class APIStats(Base): """API stats""" total: int errors: int @dataclass class AccountInfo(Base): """Account information References: * `Account Information <https://docs.nats.io/jetstream/administration/account#account-information>`_ """ # NOTE: These fields are shared with Tier type as well. memory: int storage: int streams: int consumers: int limits: AccountLimits api: APIStats domain: Optional[str] = None tiers: Optional[Dict[str, Tier]] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "limits", AccountLimits) cls._convert(resp, "api", APIStats) info = super().from_response(resp) tiers = resp.get("tiers", None) if tiers: result = {} for k, v in tiers.items(): result[k] = Tier.from_response(v) info.tiers = result return info @dataclass class RawStreamMsg(Base): subject: Optional[str] = None seq: Optional[int] = None data: Optional[bytes] = None hdrs: Optional[bytes] = None headers: Optional[Dict] = None stream: Optional[str] = None time: Optional[datetime.datetime] = None @property def sequence(self) -> Optional[int]: return self.seq @property def header(self) -> Optional[Dict]: """ header returns the headers from a message. """ return self.headers @classmethod def from_response(cls, resp: Dict[str, Any]): resp["time"] = cls._parse_utc_iso(resp["time"]) return super().from_response(resp) @dataclass class KeyValueConfig(Base): """ KeyValueConfig is the configuration of a KeyValue store. """ bucket: str description: Optional[str] = None max_value_size: Optional[int] = None history: int = 1 ttl: Optional[float] = None # in seconds max_bytes: Optional[int] = None storage: Optional[StorageType] = None replicas: int = 1 placement: Optional[Placement] = None republish: Optional[RePublish] = None direct: Optional[bool] = None def as_dict(self) -> Dict[str, object]: result = super().as_dict() result["ttl"] = self._to_nanoseconds(self.ttl) return result @dataclass class StreamPurgeRequest(Base): """ StreamPurgeRequest is optional request information to the purge API. """ # Purge up to but not including sequence. seq: Optional[int] = None # Subject to match against messages for the purge command. filter: Optional[str] = None # Number of messages to keep. keep: Optional[int] = None @dataclass class ObjectStoreConfig(Base): """ ObjectStoreConfig is the configurigation of an ObjectStore. """ bucket: Optional[str] = None description: Optional[str] = None ttl: Optional[float] = None max_bytes: Optional[int] = None storage: Optional[StorageType] = None replicas: int = 1 placement: Optional[Placement] = None def as_dict(self) -> Dict[str, object]: result = super().as_dict() result["ttl"] = self._to_nanoseconds(self.ttl) return result @dataclass class ObjectLink(Base): """ ObjectLink is used to embed links to other buckets and objects. """ # Bucket is the name of the other object store. bucket: str # Name can be used to link to a single object. # If empty means this is a link to the whole store, like a directory. name: Optional[str] = None @classmethod def from_response(cls, resp: Dict[str, Any]): return super().from_response(resp) @dataclass class ObjectMetaOptions(Base): link: Optional[ObjectLink] = None max_chunk_size: Optional[int] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "link", ObjectLink) return super().from_response(resp) @dataclass class ObjectMeta(Base): """ ObjectMeta is high level information about an object. """ name: Optional[str] = None description: Optional[str] = None headers: Optional[dict] = None # Optional options. options: Optional[ObjectMetaOptions] = None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "options", ObjectMetaOptions) return super().from_response(resp) @dataclass class ObjectInfo(Base): """ ObjectInfo is meta plus instance information. """ name: str bucket: str nuid: str size: Optional[int] = None mtime: Optional[str] = None chunks: Optional[int] = None digest: Optional[str] = None deleted: Optional[bool] = False description: Optional[str] = None headers: Optional[dict] = None # Optional options. options: Optional[ObjectMetaOptions] = None # NOTE: name, description, headers, options together compose # what would be the ObjectMeta embedded type in Go. @property def meta(self) -> ObjectMeta: return ObjectMeta( name=self.name, description=self.description, headers=self.headers, options=self.options, ) def is_link(self) -> bool: return self.options is not None and self.options.link is not None @classmethod def from_response(cls, resp: Dict[str, Any]): cls._convert(resp, "options", ObjectMetaOptions) return super().from_response(resp)