from dataclasses import dataclass, field, fields from typing import Any @dataclass class ParsedAttributes: """Base class for data structures representing parsed attributes from configs.""" @classmethod def parse(cls, attributes: dict[str, Any]): # Pick attributes supported by the class field_names = set(field.name for field in fields(cls)) supported_attributes = {key: value for key, value in attributes.items() if key in field_names} return cls(**supported_attributes) @dataclass class CLIConfig: """Describes CLI tool on some host. Attributes: name: Name of the tool. exec_path: Path to executable file of the tool. attributes: Dict with extra information about the tool. """ name: str exec_path: str attributes: dict[str, str] = field(default_factory=dict) extra_args: list[str] = field(default_factory=list) @dataclass class ServiceConfig: """Describes frostFS service on some host. Attributes: name: Name of the service that uniquely identifies it across all hosts. attributes: Dict with extra information about the service. For example, we can store name of docker container (or name of systemd service), endpoints, path to wallet, path to configuration file, etc. """ name: str attributes: dict[str, str] = field(default_factory=dict) @dataclass class HostConfig: """Describes machine that hosts frostFS services. Attributes: plugin_name: Name of plugin that should be used to manage the host. healthcheck_plugin_name: Name of the plugin for healthcheck operations. address: Address of the machine (IP or DNS name). services: List of services hosted on the machine. clis: List of CLI tools available on the machine. attributes: Dict with extra information about the host. For example, we can store connection parameters in this dict. """ plugin_name: str healthcheck_plugin_name: str address: str s3_creds_plugin_name: str = field(default="authmate") grpc_creds_plugin_name: str = field(default="wallet_factory") product: str = field(default="frostfs") services: list[ServiceConfig] = field(default_factory=list) clis: list[CLIConfig] = field(default_factory=list) attributes: dict[str, str] = field(default_factory=dict) interfaces: dict[str, str] = field(default_factory=dict) environment: dict[str, str] = field(default_factory=dict) def __post_init__(self) -> None: self.services = [ServiceConfig(**service) for service in self.services or []] self.clis = [CLIConfig(**cli) for cli in self.clis or []]