Admin

name	type	size	perms	modified	actions
[parent directory]
__pycache__/	dir	-	0755	2026-08-19 08:39:33
__init__.py	file	765 B	0644	2026-07-20 00:00:00	edit
api.py	file	27,894 B	0644	2026-07-20 00:00:00	edit
client.py	file	54,474 B	0644	2026-07-20 00:00:00	edit
errors.py	file	7,540 B	0644	2026-07-20 00:00:00	edit
kv.py	file	18,291 B	0644	2026-07-20 00:00:00	edit
manager.py	file	15,154 B	0644	2026-07-20 00:00:00	edit
object_store.py	file	17,452 B	0644	2026-07-20 00:00:00	edit

editing: manager.py

# 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 base64
import json
from email.parser import BytesParser
from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional

from nats.errors import NoRespondersError
from nats.js import api
from nats.js.errors import APIError, NotFoundError, ServiceUnavailableError

if TYPE_CHECKING:
    from nats import NATS

NATS_HDR_LINE = bytearray(b"NATS/1.0")
NATS_HDR_LINE_SIZE = len(NATS_HDR_LINE)
_CRLF_ = b"\r\n"
_CRLF_LEN_ = len(_CRLF_)

class JetStreamManager:
    """
    JetStreamManager exposes management APIs for JetStream.
    """

def __init__(
        self,
        conn: NATS,
        prefix: str = api.DEFAULT_PREFIX,
        timeout: float = 5,
    ) -> None:
        self._prefix = prefix
        self._nc = conn
        self._timeout = timeout
        self._hdr_parser = BytesParser()

async def account_info(self) -> api.AccountInfo:
        resp = await self._api_request(f"{self._prefix}.INFO", b"", timeout=self._timeout)
        return api.AccountInfo.from_response(resp)

async def find_stream_name_by_subject(self, subject: str) -> str:
        """
        Find the stream to which a subject belongs in an account.
        """

req_sub = f"{self._prefix}.STREAM.NAMES"
        req_data = json.dumps({"subject": subject})
        info = await self._api_request(req_sub, req_data.encode(), timeout=self._timeout)
        if not info["streams"]:
            raise NotFoundError
        return info["streams"][0]

async def stream_info(self, name: str, subjects_filter: Optional[str] = None) -> api.StreamInfo:
        """
        Get the latest StreamInfo by stream name.
        """
        req_data = ""
        if subjects_filter:
            req_data = json.dumps({"subjects_filter": subjects_filter})
        resp = await self._api_request(
            f"{self._prefix}.STREAM.INFO.{name}",
            req_data.encode(),
            timeout=self._timeout,
        )
        return api.StreamInfo.from_response(resp)

async def add_stream(self, config: Optional[api.StreamConfig] = None, **params) -> api.StreamInfo:
        """
        add_stream creates a stream.
        """
        if config is None:
            config = api.StreamConfig()
        config = config.evolve(**params)

stream_name = config.name
        if stream_name is None:
            raise ValueError("nats: stream name is required")

# Validate stream name
        invalid_chars = set(".*>/\\")
        has_invalid_chars = any(char in stream_name for char in invalid_chars)
        has_whitespace = any(char.isspace() for char in stream_name)
        is_not_printable = not stream_name.isprintable()

if has_invalid_chars or has_whitespace or is_not_printable:
            raise ValueError(
                f"nats: stream name ({stream_name}) is invalid. Names cannot contain whitespace, '.', '*', '>', "
                "path separators (forward or backward slash), or non-printable characters."
            )

data = json.dumps(config.as_dict())
        resp = await self._api_request(
            f"{self._prefix}.STREAM.CREATE.{stream_name}",
            data.encode(),
            timeout=self._timeout,
        )
        return api.StreamInfo.from_response(resp)

async def update_stream(self, config: Optional[api.StreamConfig] = None, **params) -> api.StreamInfo:
        """
        update_stream updates a stream.
        """
        if config is None:
            config = api.StreamConfig()
        config = config.evolve(**params)
        if config.name is None:
            raise ValueError("nats: stream name is required")

data = json.dumps(config.as_dict())
        resp = await self._api_request(
            f"{self._prefix}.STREAM.UPDATE.{config.name}",
            data.encode(),
            timeout=self._timeout,
        )
        return api.StreamInfo.from_response(resp)

async def delete_stream(self, name: str) -> bool:
        """
        Delete a stream by name.
        """
        resp = await self._api_request(f"{self._prefix}.STREAM.DELETE.{name}", timeout=self._timeout)
        return resp["success"]

async def purge_stream(
        self,
        name: str,
        seq: Optional[int] = None,
        subject: Optional[str] = None,
        keep: Optional[int] = None,
    ) -> bool:
        """
        Purge a stream by name.
        """
        stream_req: Dict[str, Any] = {}
        if seq:
            stream_req["seq"] = seq
        if subject:
            stream_req["filter"] = subject
        if keep:
            stream_req["keep"] = keep

req = json.dumps(stream_req)
        resp = await self._api_request(f"{self._prefix}.STREAM.PURGE.{name}", req.encode(), timeout=self._timeout)
        return resp["success"]

async def consumer_info(self, stream: str, consumer: str, timeout: Optional[float] = None):
        # TODO: Validate the stream and consumer names.
        if timeout is None:
            timeout = self._timeout
        resp = await self._api_request(f"{self._prefix}.CONSUMER.INFO.{stream}.{consumer}", b"", timeout=timeout)
        return api.ConsumerInfo.from_response(resp)

async def streams_info(self, offset=0) -> List[api.StreamInfo]:
        """
        streams_info retrieves a list of streams with an optional offset.
        """
        resp = await self._api_request(
            f"{self._prefix}.STREAM.LIST",
            json.dumps({"offset": offset}).encode(),
            timeout=self._timeout,
        )
        streams = []
        for stream in resp["streams"]:
            stream_info = api.StreamInfo.from_response(stream)
            streams.append(stream_info)
        return streams

async def streams_info_iterator(self, offset=0) -> Iterable[api.StreamInfo]:
        """
        streams_info retrieves a list of streams Iterator.
        """
        resp = await self._api_request(
            f"{self._prefix}.STREAM.LIST",
            json.dumps({"offset": offset}).encode(),
            timeout=self._timeout,
        )

return api.StreamsListIterator(resp["offset"], resp["total"], resp["streams"])

async def add_consumer(
        self,
        stream: str,
        config: Optional[api.ConsumerConfig] = None,
        timeout: Optional[float] = None,
        **params,
    ) -> api.ConsumerInfo:
        if not timeout:
            timeout = self._timeout
        if config is None:
            config = api.ConsumerConfig()
        config = config.evolve(**params)
        durable_name = config.durable_name
        req = {"stream_name": stream, "config": config.as_dict()}
        req_data = json.dumps(req).encode()

resp = None
        subject = ""
        version = self._nc.connected_server_version
        consumer_name_supported = version.major >= 2 and version.minor >= 9
        if consumer_name_supported and config.name:
            # NOTE: Only supported after nats-server v2.9.0
            if config.filter_subject and config.filter_subject != ">":
                subject = f"{self._prefix}.CONSUMER.CREATE.{stream}.{config.name}.{config.filter_subject}"
            else:
                subject = f"{self._prefix}.CONSUMER.CREATE.{stream}.{config.name}"
        elif durable_name:
            # NOTE: Legacy approach to create consumers. After nats-server v2.9
            # name option can be used instead.
            subject = f"{self._prefix}.CONSUMER.DURABLE.CREATE.{stream}.{durable_name}"
        else:
            subject = f"{self._prefix}.CONSUMER.CREATE.{stream}"

resp = await self._api_request(subject, req_data, timeout=timeout)
        return api.ConsumerInfo.from_response(resp)

async def delete_consumer(self, stream: str, consumer: str) -> bool:
        resp = await self._api_request(
            f"{self._prefix}.CONSUMER.DELETE.{stream}.{consumer}",
            b"",
            timeout=self._timeout,
        )
        return resp["success"]

async def pause_consumer(
        self,
        stream: str,
        consumer: str,
        pause_until: str,
        timeout: Optional[float] = None,
    ) -> api.ConsumerPause:
        """
        Pause a consumer until the specified time.

Args:
            stream: The stream name
            consumer: The consumer name
            pause_until: RFC 3339 timestamp string (e.g., "2025-10-22T12:00:00Z")
                        until which the consumer should be paused
            timeout: Request timeout in seconds

Returns:
            ConsumerPause with paused status

Note:
            Requires nats-server 2.11.0 or later
        """
        if timeout is None:
            timeout = self._timeout

req = {"pause_until": pause_until}
        req_data = json.dumps(req).encode()

resp = await self._api_request(
            f"{self._prefix}.CONSUMER.PAUSE.{stream}.{consumer}",
            req_data,
            timeout=timeout,
        )
        return api.ConsumerPause.from_response(resp)

async def resume_consumer(
        self,
        stream: str,
        consumer: str,
        timeout: Optional[float] = None,
    ) -> api.ConsumerPause:
        """
        Resume a paused consumer immediately.

This is equivalent to calling pause_consumer with a timestamp in the past.

Args:
            stream: The stream name
            consumer: The consumer name
            timeout: Request timeout in seconds

Returns:
            ConsumerPause with paused=False

Note:
            Requires nats-server 2.11.0 or later
        """
        # Resume by pausing until a time in the past (epoch)
        return await self.pause_consumer(stream, consumer, "1970-01-01T00:00:00Z", timeout)

async def consumers_info(self, stream: str, offset: Optional[int] = None) -> List[api.ConsumerInfo]:
        """
        consumers_info retrieves a list of consumers. Consumers list limit is 256 for more
        consider to use offset
        :param stream: stream to get consumers
        :param offset: consumers list offset
        """
        resp = await self._api_request(
            f"{self._prefix}.CONSUMER.LIST.{stream}",
            b"" if offset is None else json.dumps({"offset": offset}).encode(),
            timeout=self._timeout,
        )
        consumers = []
        for consumer in resp["consumers"]:
            consumer_info = api.ConsumerInfo.from_response(consumer)
            consumers.append(consumer_info)
        return consumers

async def get_msg(
        self,
        stream_name: str,
        seq: Optional[int] = None,
        subject: Optional[str] = None,
        direct: Optional[bool] = False,
        next: Optional[bool] = False,
    ) -> api.RawStreamMsg:
        """
        get_msg retrieves a message from a stream.
        """
        req_subject = None
        req: Dict[str, Any] = {}
        if seq:
            req["seq"] = seq
        if subject:
            req["seq"] = None
            req.pop("seq", None)
            req["last_by_subj"] = subject
        if next:
            req["seq"] = seq
            req["last_by_subj"] = None
            req.pop("last_by_subj", None)
            req["next_by_subj"] = subject
        data = json.dumps(req)

if direct:
            # $JS.API.DIRECT.GET.KV_{stream_name}.$KV.TEST.{key}
            if subject and (seq is None):
                # last_by_subject type request requires no payload.
                data = ""
                req_subject = f"{self._prefix}.DIRECT.GET.{stream_name}.{subject}"
            else:
                req_subject = f"{self._prefix}.DIRECT.GET.{stream_name}"

resp = await self._nc.request(req_subject, data.encode(), timeout=self._timeout)
            raw_msg = JetStreamManager._lift_msg_to_raw_msg(resp)
            return raw_msg

# Non Direct form
        req_subject = f"{self._prefix}.STREAM.MSG.GET.{stream_name}"
        resp_data = await self._api_request(req_subject, data.encode(), timeout=self._timeout)

raw_msg = api.RawStreamMsg.from_response(resp_data["message"])
        if raw_msg.hdrs:
            hdrs = base64.b64decode(raw_msg.hdrs)
            raw_headers = hdrs[NATS_HDR_LINE_SIZE + _CRLF_LEN_ :]
            parsed_headers = self._hdr_parser.parsebytes(raw_headers)
            headers = None
            if len(parsed_headers.items()) > 0:
                headers = {}
                for k, v in parsed_headers.items():
                    headers[k] = v
            raw_msg.headers = headers

msg_data: Optional[bytes] = None
        if raw_msg.data:
            msg_data = base64.b64decode(raw_msg.data)
        raw_msg.data = msg_data

return raw_msg

@classmethod
    def _lift_msg_to_raw_msg(self, msg) -> api.RawStreamMsg:
        if not msg.data:
            msg.data = None
            status = msg.headers.get("Status")
            if status:
                if status == "404":
                    raise NotFoundError
                else:
                    raise APIError.from_msg(msg)

raw_msg = api.RawStreamMsg()
        subject = msg.headers["Nats-Subject"]
        raw_msg.subject = subject

seq = msg.headers.get("Nats-Sequence")
        if seq:
            raw_msg.seq = int(seq)
        raw_msg.data = msg.data
        raw_msg.headers = msg.headers

return raw_msg

async def delete_msg(self, stream_name: str, seq: int) -> bool:
        """
        delete_msg retrieves a message from a stream based on the sequence ID.
        """
        req_subject = f"{self._prefix}.STREAM.MSG.DELETE.{stream_name}"
        req = {"seq": seq}
        data = json.dumps(req)
        resp = await self._api_request(req_subject, data.encode())
        return resp["success"]

async def get_last_msg(
        self,
        stream_name: str,
        subject: str,
        direct: Optional[bool] = False,
    ) -> api.RawStreamMsg:
        """
        get_last_msg retrieves the last message from a stream.
        """
        return await self.get_msg(stream_name, subject=subject, direct=direct)

async def _api_request(
        self,
        req_subject: str,
        req: bytes = b"",
        timeout: float = 5,
    ) -> Dict[str, Any]:
        try:
            msg = await self._nc.request(req_subject, req, timeout=timeout)
            resp = json.loads(msg.data)
        except NoRespondersError:
            raise ServiceUnavailableError

# Check for API errors.
        if "error" in resp:
            raise APIError.from_error(resp["error"])

return resp

cancel