Sid Gifari Advanced File Manager

✏️ Editing: api.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 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)

📁 File Manager Pro