in elastic_transport/client_utils.py [0:0]
def parse_cloud_id(cloud_id: str) -> CloudId:
"""Parses an Elastic Cloud ID into its components"""
try:
cloud_id = to_str(cloud_id)
cluster_name, _, cloud_id = cloud_id.partition(":")
parts = to_str(binascii.a2b_base64(to_bytes(cloud_id, "ascii")), "ascii").split(
"$"
)
parent_dn = parts[0]
if not parent_dn:
raise ValueError() # Caught and re-raised properly below
es_uuid: Optional[str]
kibana_uuid: Optional[str]
try:
es_uuid = parts[1]
except IndexError:
es_uuid = None
try:
kibana_uuid = parts[2] or None
except IndexError:
kibana_uuid = None
if ":" in parent_dn:
parent_dn, _, parent_port = parent_dn.rpartition(":")
port = int(parent_port)
else:
port = 443
except (ValueError, IndexError, UnicodeError):
raise ValueError("Cloud ID is not properly formatted") from None
es_host = f"{es_uuid}.{parent_dn}" if es_uuid else None
kibana_host = f"{kibana_uuid}.{parent_dn}" if kibana_uuid else None
return CloudId(
cluster_name=cluster_name,
es_address=(es_host, port) if es_host else None,
kibana_address=(kibana_host, port) if kibana_host else None,
)