import datetime
from typing import Any, Dict, List, Type, TypeVar, Union
import attr
from dateutil.parser import isoparse
from ..models.cluster import Cluster
from ..models.gateway import Gateway
from ..models.jetstream import Jetstream
from ..models.leaf_node import LeafNode
from ..types import UNSET, Unset
ON = TypeVar("ON", bound="Connection")
[docs]@attr.s(auto_attribs=True)
class Connection:
"""Metadata about a pub-sub connection.
This is mostly used for internal purposes and debugging.""" # noqa: E501
auth_timeout: Union[Unset, int] = UNSET
cluster: Union[Unset, Cluster] = UNSET
config_load_time: Union[Unset, datetime.datetime] = UNSET
connections: Union[Unset, int] = UNSET
cores: Union[Unset, int] = UNSET
cpu: Union[Unset, float] = UNSET
gateway: Union[Unset, Gateway] = UNSET
git_commit: Union[Unset, str] = UNSET
go: Union[Unset, str] = UNSET
gomaxprocs: Union[Unset, int] = UNSET
host: Union[Unset, str] = UNSET
http_base_path: Union[Unset, str] = UNSET
http_host: Union[Unset, str] = UNSET
http_port: Union[Unset, int] = UNSET
http_req_stats: Union[Unset, Any] = UNSET
https_port: Union[Unset, int] = UNSET
in_bytes: Union[Unset, int] = UNSET
in_msgs: Union[Unset, int] = UNSET
jetstream: Union[Unset, Jetstream] = UNSET
leaf: Union[Unset, LeafNode] = UNSET
leafnodes: Union[Unset, int] = UNSET
max_connections: Union[Unset, int] = UNSET
max_control_line: Union[Unset, int] = UNSET
max_payload: Union[Unset, int] = UNSET
max_pending: Union[Unset, int] = UNSET
mem: Union[Unset, int] = UNSET
now: Union[Unset, datetime.datetime] = UNSET
out_bytes: Union[Unset, int] = UNSET
out_msgs: Union[Unset, int] = UNSET
ping_interval: Union[Unset, int] = UNSET
ping_max: Union[Unset, int] = UNSET
port: Union[Unset, int] = UNSET
proto: Union[Unset, int] = UNSET
remotes: Union[Unset, int] = UNSET
routes: Union[Unset, int] = UNSET
server_id: Union[Unset, str] = UNSET
server_name: Union[Unset, str] = UNSET
slow_consumers: Union[Unset, int] = UNSET
start: Union[Unset, datetime.datetime] = UNSET
subscriptions: Union[Unset, int] = UNSET
system_account: Union[Unset, str] = UNSET
tls_timeout: Union[Unset, int] = UNSET
total_connections: Union[Unset, int] = UNSET
uptime: Union[Unset, str] = UNSET
version: Union[Unset, str] = UNSET
write_deadline: Union[Unset, int] = UNSET
additional_properties: Dict[str, Any] = attr.ib(init=False, factory=dict)
[docs] def to_dict(self) -> Dict[str, Any]:
auth_timeout = self.auth_timeout
if not isinstance(self.cluster, Unset):
cluster = self.cluster
config_load_time: Union[Unset, str] = UNSET
if not isinstance(self.config_load_time, Unset):
config_load_time = self.config_load_time.isoformat()
connections = self.connections
cores = self.cores
cpu = self.cpu
if not isinstance(self.gateway, Unset):
gateway = self.gateway
git_commit = self.git_commit
go = self.go
gomaxprocs = self.gomaxprocs
host = self.host
http_base_path = self.http_base_path
http_host = self.http_host
http_port = self.http_port
http_req_stats = self.http_req_stats
https_port = self.https_port
in_bytes = self.in_bytes
in_msgs = self.in_msgs
if not isinstance(self.jetstream, Unset):
jetstream = self.jetstream
if not isinstance(self.leaf, Unset):
leaf = self.leaf
leafnodes = self.leafnodes
max_connections = self.max_connections
max_control_line = self.max_control_line
max_payload = self.max_payload
max_pending = self.max_pending
mem = self.mem
now: Union[Unset, str] = UNSET
if not isinstance(self.now, Unset):
now = self.now.isoformat()
out_bytes = self.out_bytes
out_msgs = self.out_msgs
ping_interval = self.ping_interval
ping_max = self.ping_max
port = self.port
proto = self.proto
remotes = self.remotes
routes = self.routes
server_id = self.server_id
server_name = self.server_name
slow_consumers = self.slow_consumers
start: Union[Unset, str] = UNSET
if not isinstance(self.start, Unset):
start = self.start.isoformat()
subscriptions = self.subscriptions
system_account = self.system_account
tls_timeout = self.tls_timeout
total_connections = self.total_connections
uptime = self.uptime
version = self.version
write_deadline = self.write_deadline
field_dict: Dict[str, Any] = {}
field_dict.update(self.additional_properties)
field_dict.update({})
if auth_timeout is not UNSET:
field_dict["auth_timeout"] = auth_timeout
if cluster is not UNSET:
field_dict["cluster"] = cluster
if config_load_time is not UNSET:
field_dict["config_load_time"] = config_load_time
if connections is not UNSET:
field_dict["connections"] = connections
if cores is not UNSET:
field_dict["cores"] = cores
if cpu is not UNSET:
field_dict["cpu"] = cpu
if gateway is not UNSET:
field_dict["gateway"] = gateway
if git_commit is not UNSET:
field_dict["git_commit"] = git_commit
if go is not UNSET:
field_dict["go"] = go
if gomaxprocs is not UNSET:
field_dict["gomaxprocs"] = gomaxprocs
if host is not UNSET:
field_dict["host"] = host
if http_base_path is not UNSET:
field_dict["http_base_path"] = http_base_path
if http_host is not UNSET:
field_dict["http_host"] = http_host
if http_port is not UNSET:
field_dict["http_port"] = http_port
if http_req_stats is not UNSET:
field_dict["http_req_stats"] = http_req_stats
if https_port is not UNSET:
field_dict["https_port"] = https_port
if in_bytes is not UNSET:
field_dict["in_bytes"] = in_bytes
if in_msgs is not UNSET:
field_dict["in_msgs"] = in_msgs
if jetstream is not UNSET:
field_dict["jetstream"] = jetstream
if leaf is not UNSET:
field_dict["leaf"] = leaf
if leafnodes is not UNSET:
field_dict["leafnodes"] = leafnodes
if max_connections is not UNSET:
field_dict["max_connections"] = max_connections
if max_control_line is not UNSET:
field_dict["max_control_line"] = max_control_line
if max_payload is not UNSET:
field_dict["max_payload"] = max_payload
if max_pending is not UNSET:
field_dict["max_pending"] = max_pending
if mem is not UNSET:
field_dict["mem"] = mem
if now is not UNSET:
field_dict["now"] = now
if out_bytes is not UNSET:
field_dict["out_bytes"] = out_bytes
if out_msgs is not UNSET:
field_dict["out_msgs"] = out_msgs
if ping_interval is not UNSET:
field_dict["ping_interval"] = ping_interval
if ping_max is not UNSET:
field_dict["ping_max"] = ping_max
if port is not UNSET:
field_dict["port"] = port
if proto is not UNSET:
field_dict["proto"] = proto
if remotes is not UNSET:
field_dict["remotes"] = remotes
if routes is not UNSET:
field_dict["routes"] = routes
if server_id is not UNSET:
field_dict["server_id"] = server_id
if server_name is not UNSET:
field_dict["server_name"] = server_name
if slow_consumers is not UNSET:
field_dict["slow_consumers"] = slow_consumers
if start is not UNSET:
field_dict["start"] = start
if subscriptions is not UNSET:
field_dict["subscriptions"] = subscriptions
if system_account is not UNSET:
field_dict["system_account"] = system_account
if tls_timeout is not UNSET:
field_dict["tls_timeout"] = tls_timeout
if total_connections is not UNSET:
field_dict["total_connections"] = total_connections
if uptime is not UNSET:
field_dict["uptime"] = uptime
if version is not UNSET:
field_dict["version"] = version
if write_deadline is not UNSET:
field_dict["write_deadline"] = write_deadline
return field_dict
[docs] @classmethod
def from_dict(cls: Type[ON], src_dict: Dict[str, Any]) -> ON:
d = src_dict.copy()
auth_timeout = d.pop("auth_timeout", UNSET)
_cluster = d.pop("cluster", UNSET)
cluster: Union[Unset, Cluster]
if isinstance(_cluster, Unset):
cluster = UNSET
else:
cluster = _cluster # type: ignore[arg-type]
_config_load_time = d.pop("config_load_time", UNSET)
config_load_time: Union[Unset, datetime.datetime]
if isinstance(_config_load_time, Unset):
config_load_time = UNSET
else:
config_load_time = isoparse(_config_load_time)
connections = d.pop("connections", UNSET)
cores = d.pop("cores", UNSET)
cpu = d.pop("cpu", UNSET)
_gateway = d.pop("gateway", UNSET)
gateway: Union[Unset, Gateway]
if isinstance(_gateway, Unset):
gateway = UNSET
else:
gateway = _gateway # type: ignore[arg-type]
git_commit = d.pop("git_commit", UNSET)
go = d.pop("go", UNSET)
gomaxprocs = d.pop("gomaxprocs", UNSET)
host = d.pop("host", UNSET)
http_base_path = d.pop("http_base_path", UNSET)
http_host = d.pop("http_host", UNSET)
http_port = d.pop("http_port", UNSET)
http_req_stats = d.pop("http_req_stats", UNSET)
https_port = d.pop("https_port", UNSET)
in_bytes = d.pop("in_bytes", UNSET)
in_msgs = d.pop("in_msgs", UNSET)
_jetstream = d.pop("jetstream", UNSET)
jetstream: Union[Unset, Jetstream]
if isinstance(_jetstream, Unset):
jetstream = UNSET
else:
jetstream = _jetstream # type: ignore[arg-type]
_leaf = d.pop("leaf", UNSET)
leaf: Union[Unset, LeafNode]
if isinstance(_leaf, Unset):
leaf = UNSET
else:
leaf = _leaf # type: ignore[arg-type]
leafnodes = d.pop("leafnodes", UNSET)
max_connections = d.pop("max_connections", UNSET)
max_control_line = d.pop("max_control_line", UNSET)
max_payload = d.pop("max_payload", UNSET)
max_pending = d.pop("max_pending", UNSET)
mem = d.pop("mem", UNSET)
_now = d.pop("now", UNSET)
now: Union[Unset, datetime.datetime]
if isinstance(_now, Unset):
now = UNSET
else:
now = isoparse(_now)
out_bytes = d.pop("out_bytes", UNSET)
out_msgs = d.pop("out_msgs", UNSET)
ping_interval = d.pop("ping_interval", UNSET)
ping_max = d.pop("ping_max", UNSET)
port = d.pop("port", UNSET)
proto = d.pop("proto", UNSET)
remotes = d.pop("remotes", UNSET)
routes = d.pop("routes", UNSET)
server_id = d.pop("server_id", UNSET)
server_name = d.pop("server_name", UNSET)
slow_consumers = d.pop("slow_consumers", UNSET)
_start = d.pop("start", UNSET)
start: Union[Unset, datetime.datetime]
if isinstance(_start, Unset):
start = UNSET
else:
start = isoparse(_start)
subscriptions = d.pop("subscriptions", UNSET)
system_account = d.pop("system_account", UNSET)
tls_timeout = d.pop("tls_timeout", UNSET)
total_connections = d.pop("total_connections", UNSET)
uptime = d.pop("uptime", UNSET)
version = d.pop("version", UNSET)
write_deadline = d.pop("write_deadline", UNSET)
connection = cls(
auth_timeout=auth_timeout,
cluster=cluster,
config_load_time=config_load_time,
connections=connections,
cores=cores,
cpu=cpu,
gateway=gateway,
git_commit=git_commit,
go=go,
gomaxprocs=gomaxprocs,
host=host,
http_base_path=http_base_path,
http_host=http_host,
http_port=http_port,
http_req_stats=http_req_stats,
https_port=https_port,
in_bytes=in_bytes,
in_msgs=in_msgs,
jetstream=jetstream,
leaf=leaf,
leafnodes=leafnodes,
max_connections=max_connections,
max_control_line=max_control_line,
max_payload=max_payload,
max_pending=max_pending,
mem=mem,
now=now,
out_bytes=out_bytes,
out_msgs=out_msgs,
ping_interval=ping_interval,
ping_max=ping_max,
port=port,
proto=proto,
remotes=remotes,
routes=routes,
server_id=server_id,
server_name=server_name,
slow_consumers=slow_consumers,
start=start,
subscriptions=subscriptions,
system_account=system_account,
tls_timeout=tls_timeout,
total_connections=total_connections,
uptime=uptime,
version=version,
write_deadline=write_deadline,
)
connection.additional_properties = d
return connection
@property
def additional_keys(self) -> List[str]:
return list(self.additional_properties.keys())
[docs] def __getitem__(self, key: str) -> Any:
return self.additional_properties[key]
[docs] def __setitem__(self, key: str, value: Any) -> None:
self.additional_properties[key] = value
[docs] def __delitem__(self, key: str) -> None:
del self.additional_properties[key]
[docs] def __contains__(self, key: str) -> bool:
return key in self.additional_properties