"""YAML-based chain credentials, ported from ``credentials/credentials.go``. The YAML format mirrors the Go SDK:: default: my-public-id chains: - name: my-chain publicId: my-public-id authKeyId: my-auth-key-id authKey: my-auth-key endpoint: https://chains.dragonchain.com Note: unlike the Go version (which uses yaml.Node to preserve comments and blank-line formatting), this loader uses PyYAML and does not round-trip comments on ``save_config``. """ import os from dataclasses import dataclass, field from typing import List, Optional import yaml @dataclass class ChainConfig: name: str = "" public_id: str = "" auth_key_id: str = "" auth_key: str = "" endpoint: str = "" @classmethod def from_dict(cls, d: dict) -> "ChainConfig": return cls( name=d.get("name", ""), public_id=d.get("publicId", ""), auth_key_id=d.get("authKeyId", ""), auth_key=d.get("authKey", ""), endpoint=d.get("endpoint", ""), ) def to_dict(self) -> dict: return { "name": self.name, "publicId": self.public_id, "authKeyId": self.auth_key_id, "authKey": self.auth_key, "endpoint": self.endpoint, } @dataclass class Config: default: str = "" chains: List[ChainConfig] = field(default_factory=list) # -- factory helpers ---------------------------------------------------- @classmethod def from_dict(cls, d: dict) -> "Config": return cls( default=d.get("default", ""), chains=[ChainConfig.from_dict(c) for c in (d.get("chains") or [])], ) def to_dict(self) -> dict: return { "default": self.default, "chains": [c.to_dict() for c in self.chains], } # -- lookups ------------------------------------------------------------ def get_default_chain(self) -> Optional[ChainConfig]: """Return the chain configuration for the default publicId, or None.""" return self.get_chain_by_public_id(self.default) def get_chain_by_public_id(self, public_id: str) -> Optional[ChainConfig]: for chain in self.chains: if chain.public_id == public_id: return chain return None def list_chains(self) -> List[str]: """Return all chain names.""" return [c.name for c in self.chains] # -- mutations ---------------------------------------------------------- def add_chain(self, chain: ChainConfig) -> None: self.chains.append(chain) def set_default(self, public_id: str) -> None: self.default = public_id def delete_chain(self, public_id: str) -> None: """Delete a chain by publicId. Raises ValueError if it's the default.""" if public_id == self.default: raise ValueError("cannot delete default chain") self.chains = [c for c in self.chains if c.public_id != public_id] # -- persistence -------------------------------------------------------- def save_config(self, file_path: str) -> None: expanded = expand_path(file_path) with open(expanded, "w", encoding="utf-8") as f: yaml.safe_dump(self.to_dict(), f, sort_keys=False, default_flow_style=False) def load_config(file_path: str) -> Config: """Read and parse a YAML configuration file.""" expanded = expand_path(file_path) with open(expanded, "r", encoding="utf-8") as f: data = yaml.safe_load(f) or {} return Config.from_dict(data) def load_config_from_string(yaml_content: str) -> Config: """Parse YAML configuration from a string.""" data = yaml.safe_load(yaml_content) or {} return Config.from_dict(data) def expand_path(path: str) -> str: """Expand environment variables and a leading ``~`` in a path.""" return os.path.expanduser(os.path.expandvars(path))