From 0a9cfb33389edf6e671e9d527b5d4d7071b11bc6 Mon Sep 17 00:00:00 2001 From: Teknique Date: Fri, 16 Jun 2023 18:17:17 -0700 Subject: [PATCH] Use dataclasses for configuration. --- veilid-python/veilid/config.py | 472 +++++---------------------------- 1 file changed, 72 insertions(+), 400 deletions(-) diff --git a/veilid-python/veilid/config.py b/veilid-python/veilid/config.py index d9394c4e..59522c50 100644 --- a/veilid-python/veilid/config.py +++ b/veilid-python/veilid/config.py @@ -1,18 +1,46 @@ -from typing import Self, Optional +from dataclasses import dataclass, fields from enum import StrEnum -from json import dumps +from typing import Optional, Self + +from .types import TypedKey, TypedSecret -from .types import * class VeilidConfigLogLevel(StrEnum): - OFF = 'Off' - ERROR = 'Error' - WARN = 'Warn' - INFO = 'Info' - DEBUG = 'Debug' - TRACE = 'Trace' + OFF = "Off" + ERROR = "Error" + WARN = "Warn" + INFO = "Info" + DEBUG = "Debug" + TRACE = "Trace" -class VeilidConfigCapabilities: + +@dataclass +class ConfigBase: + @classmethod + def from_json(cls, json_data: dict) -> Self: + """Return an instance of this type from the input data.""" + args = {} + for field in fields(cls): + key = field.name + value = json_data[key] + try: + # See if this field's type knows how to load itself from JSON input. + loader = field.type.from_json + except AttributeError: + # No, it doesn't. Use the raw value. + args[key] = value + else: + # Yes, it does. Use the loading function's output. + args[key] = loader(value) + + return cls(**args) + + def to_json(self) -> dict: + return self.__dict__ + + +@dataclass +class VeilidConfigCapabilities(ConfigBase): protocol_udp: bool protocol_connect_tcp: bool protocol_accept_tcp: bool @@ -21,30 +49,9 @@ class VeilidConfigCapabilities: protocol_connect_wss: bool protocol_accept_wss: bool - def __init__(self, protocol_udp: bool, protocol_connect_tcp: bool, protocol_accept_tcp: bool, - protocol_connect_ws: bool, protocol_accept_ws: bool, protocol_connect_wss: bool, protocol_accept_wss: bool): - self.protocol_udp = protocol_udp - self.protocol_connect_tcp = protocol_connect_tcp - self.protocol_accept_tcp = protocol_accept_tcp - self.protocol_connect_ws = protocol_connect_ws - self.protocol_accept_ws = protocol_accept_ws - self.protocol_connect_wss = protocol_connect_wss - self.protocol_accept_wss = protocol_accept_wss - - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigCapabilities(j['protocol_udp'], - j['protocol_connect_tcp'], - j['protocol_accept_tcp'], - j['protocol_connect_ws'], - j['protocol_accept_ws'], - j['protocol_connect_wss'], - j['protocol_accept_wss']) - def to_json(self) -> dict: - return self.__dict__ - -class VeilidConfigProtectedStore: +@dataclass +class VeilidConfigProtectedStore(ConfigBase): allow_insecure_fallback: bool always_use_insecure_storage: bool directory: str @@ -52,52 +59,21 @@ class VeilidConfigProtectedStore: device_encryption_key_password: str new_device_encryption_key_password: Optional[str] - def __init__(self, allow_insecure_fallback: bool, always_use_insecure_storage: bool, - directory: str, delete: bool, device_encryption_key_password: str, new_device_encryption_key_password: Optional[str]): - self.allow_insecure_fallback = allow_insecure_fallback - self.always_use_insecure_storage = always_use_insecure_storage - self.directory = directory - self.delete = delete - self.device_encryption_key_password = device_encryption_key_password - self.new_device_encryption_key_password = new_device_encryption_key_password - - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigProtectedStore(j['allow_insecure_fallback'], j['always_use_insecure_storage'], - j['directory'], j['delete'], j['device_encryption_key_password'], j['new_device_encryption_key_password']) - def to_json(self) -> dict: - return self.__dict__ - -class VeilidConfigTableStore: +@dataclass +class VeilidConfigTableStore(ConfigBase): directory: str delete: bool - def __init__(self, directory: str, delete: bool): - self.directory = directory - self.delete = delete - - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigTableStore(j['directory'], j['delete']) - def to_json(self) -> dict: - return self.__dict__ -class VeilidConfigBlockStore: +@dataclass +class VeilidConfigBlockStore(ConfigBase): directory: str delete: bool - def __init__(self, directory: str, delete: bool): - self.directory = directory - self.delete = delete - - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigBlockStore(j['directory'], j['delete']) - def to_json(self) -> dict: - return self.__dict__ -class VeilidConfigRoutingTable: +@dataclass +class VeilidConfigRoutingTable(ConfigBase): node_id: list[TypedKey] node_id_secret: list[TypedSecret] bootstrap: list[str] @@ -107,34 +83,9 @@ class VeilidConfigRoutingTable: limit_attached_good: int limit_attached_weak: int - def __init__(self, node_id: list[TypedKey], node_id_secret: list[TypedSecret], bootstrap: list[str], limit_over_attached: int, - limit_fully_attached: int, limit_attached_strong: int, limit_attached_good: int, limit_attached_weak: int): - self.node_id = node_id - self.node_id_secret = node_id_secret - self.bootstrap = bootstrap - self.limit_over_attached = limit_over_attached - self.limit_fully_attached = limit_fully_attached - self.limit_attached_strong = limit_attached_strong - self.limit_attached_good = limit_attached_good - self.limit_attached_weak = limit_attached_weak - - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigRoutingTable( - list(map(lambda x: TypedKey(x), j['node_id'])), - list(map(lambda x: TypedSecret(x), j['node_id_secret'])), - j['bootstrap'], - j['limit_over_attached'], - j['limit_fully_attached'], - j['limit_attached_strong'], - j['limit_attached_good'], - j['limit_attached_weak']) - def to_json(self) -> dict: - return self.__dict__ - - -class VeilidConfigRPC: +@dataclass +class VeilidConfigRPC(ConfigBase): concurrency: int queue_size: int max_timestamp_behind_ms: Optional[int] @@ -143,31 +94,9 @@ class VeilidConfigRPC: max_route_hop_count: int default_route_hop_count: int - def __init__(self, concurrency: int, queue_size: int, max_timestamp_behind_ms: Optional[int], max_timestamp_ahead_ms: Optional[int], - timeout_ms: int, max_route_hop_count: int, default_route_hop_count: int): - self.concurrency = concurrency - self.queue_size = queue_size - self.max_timestamp_behind_ms = max_timestamp_behind_ms - self.max_timestamp_ahead_ms = max_timestamp_ahead_ms - self.timeout_ms = timeout_ms - self.max_route_hop_count = max_route_hop_count - self.default_route_hop_count = default_route_hop_count - - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigRPC( - j['concurrency'], - j['queue_size'], - j['max_timestamp_behind_ms'], - j['max_timestamp_ahead_ms'], - j['timeout_ms'], - j['max_route_hop_count'], - j['default_route_hop_count']) - def to_json(self) -> dict: - return self.__dict__ - -class VeilidConfigDHT: +@dataclass +class VeilidConfigDHT(ConfigBase): max_find_node_count: int resolve_node_timeout_ms: int resolve_node_count: int @@ -188,265 +117,83 @@ class VeilidConfigDHT: remote_max_subkey_cache_memory_mb: int remote_max_storage_space_mb: int - def __init__(self, max_find_node_count: int, resolve_node_timeout_ms: int, resolve_node_count: int, - resolve_node_fanout: int, get_value_timeout_ms: int, get_value_count: int, get_value_fanout: int, - set_value_timeout_ms: int, set_value_count: int, set_value_fanout: int, - min_peer_count: int, min_peer_refresh_time_ms: int, validate_dial_info_receipt_time_ms: int, - local_subkey_cache_size: int, local_max_subkey_cache_memory_mb: int, - remote_subkey_cache_size: int, remote_max_records: int, remote_max_subkey_cache_memory_mb: int, remote_max_storage_space_mb: int): - self.max_find_node_count = max_find_node_count - self.resolve_node_timeout_ms =resolve_node_timeout_ms - self.resolve_node_count = resolve_node_count - self.resolve_node_fanout = resolve_node_fanout - self.get_value_timeout_ms = get_value_timeout_ms - self.get_value_count = get_value_count - self.get_value_fanout = get_value_fanout - self.set_value_timeout_ms = set_value_timeout_ms - self.set_value_count = set_value_count - self.set_value_fanout = set_value_fanout - self.min_peer_count = min_peer_count - self.min_peer_refresh_time_ms = min_peer_refresh_time_ms - self.validate_dial_info_receipt_time_ms = validate_dial_info_receipt_time_ms - self.local_subkey_cache_size = local_subkey_cache_size - self.local_max_subkey_cache_memory_mb = local_max_subkey_cache_memory_mb - self.remote_subkey_cache_size = remote_subkey_cache_size - self.remote_max_records = remote_max_records - self.remote_max_subkey_cache_memory_mb = remote_max_subkey_cache_memory_mb - self.remote_max_storage_space_mb = remote_max_storage_space_mb - - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigDHT( - j['max_find_node_count'], - j['resolve_node_timeout_ms'], - j['resolve_node_count'], - j['resolve_node_fanout'], - j['get_value_timeout_ms'], - j['get_value_count'], - j['get_value_fanout'], - j['set_value_timeout_ms'], - j['set_value_count'], - j['set_value_fanout'], - j['min_peer_count'], - j['min_peer_refresh_time_ms'], - j['validate_dial_info_receipt_time_ms'], - j['local_subkey_cache_size'], - j['local_max_subkey_cache_memory_mb'], - j['remote_subkey_cache_size'], - j['remote_max_records'], - j['remote_max_subkey_cache_memory_mb'], - j['remote_max_storage_space_mb']) - def to_json(self) -> dict: - return self.__dict__ - -class VeilidConfigTLS: +@dataclass +class VeilidConfigTLS(ConfigBase): certificate_path: str private_key_path: str connection_initial_timeout_ms: int - def __init__(self, certificate_path: str, private_key_path: str, connection_initial_timeout_ms: int): - self.certificate_path = certificate_path - self.private_key_path = private_key_path - self.connection_initial_timeout_ms = connection_initial_timeout_ms - - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigTLS( - j['certificate_path'], - j['private_key_path'], - j['connection_initial_timeout_ms']) - def to_json(self) -> dict: - return self.__dict__ -class VeilidConfigHTTPS: +@dataclass +class VeilidConfigHTTPS(ConfigBase): enabled: bool listen_address: str path: str url: Optional[str] - - def __init__(self, enabled: bool, listen_address: str, path: str, url: Optional[str]): - self.enabled = enabled - self.listen_address = listen_address - self.path = path - self.url = url - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigHTTPS( - j['enabled'], - j['listen_address'], - j['path'], - j['url']) - def to_json(self) -> dict: - return self.__dict__ -class VeilidConfigHTTP: +@dataclass +class VeilidConfigHTTP(ConfigBase): enabled: bool listen_address: str path: str url: Optional[str] - - def __init__(self, enabled: bool, listen_address: str, path: str, url: Optional[str]): - self.enabled = enabled - self.listen_address = listen_address - self.path = path - self.url = url - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigHTTP( - j['enabled'], - j['listen_address'], - j['path'], - j['url']) - def to_json(self) -> dict: - return self.__dict__ -class VeilidConfigApplication: +@dataclass +class VeilidConfigApplication(ConfigBase): https: VeilidConfigHTTPS http: VeilidConfigHTTP - def __init__(self, https: VeilidConfigHTTPS, http: VeilidConfigHTTP): - self.https = https - self.http = http - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigApplication( - VeilidConfigHTTPS.from_json(j['https']), - VeilidConfigHTTP.from_json(j['http'])) - def to_json(self) -> dict: - return self.__dict__ - - -class VeilidConfigUDP: +@dataclass +class VeilidConfigUDP(ConfigBase): enabled: bool socket_pool_size: int listen_address: str public_address: Optional[str] - - def __init__(self, enabled: bool, socket_pool_size: int, listen_address: str, public_address: Optional[str]): - self.enabled = enabled - self.socket_pool_size = socket_pool_size - self.listen_address = listen_address - self.public_address = public_address - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigUDP( - j['enabled'], - j['socket_pool_size'], - j['listen_address'], - j['public_address']) - def to_json(self) -> dict: - return self.__dict__ -class VeilidConfigTCP: +@dataclass +class VeilidConfigTCP(ConfigBase): connect: bool listen: bool max_connections: int listen_address: str public_address: Optional[str] - - def __init__(self, connect: bool, listen: bool, max_connections: int, listen_address: str, public_address: Optional[str]): - self.connect = connect - self.listen = listen - self.max_connections = max_connections - self.listen_address = listen_address - self.public_address = public_address - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigTCP( - j['connect'], - j['listen'], - j['max_connections'], - j['listen_address'], - j['public_address']) - def to_json(self) -> dict: - return self.__dict__ -class VeilidConfigWS: +@dataclass +class VeilidConfigWS(ConfigBase): connect: bool listen: bool max_connections: int listen_address: str path: str url: Optional[str] - - def __init__(self, connect: bool, listen: bool, max_connections: int, listen_address: str, path: str, url: Optional[str]): - self.connect = connect - self.listen = listen - self.max_connections = max_connections - self.listen_address = listen_address - self.path = path - self.url = url - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigWS( - j['connect'], - j['listen'], - j['max_connections'], - j['listen_address'], - j['path'], - j['url']) - def to_json(self) -> dict: - return self.__dict__ -class VeilidConfigWSS: +@dataclass +class VeilidConfigWSS(ConfigBase): connect: bool listen: bool max_connections: int listen_address: str path: str url: Optional[str] - - def __init__(self, connect: bool, listen: bool, max_connections: int, listen_address: str, path: str, url: Optional[str]): - self.connect = connect - self.listen = listen - self.max_connections = max_connections - self.listen_address = listen_address - self.path = path - self.url = url - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigWSS( - j['connect'], - j['listen'], - j['max_connections'], - j['listen_address'], - j['path'], - j['url']) - def to_json(self) -> dict: - return self.__dict__ -class VeilidConfigProtocol: +@dataclass +class VeilidConfigProtocol(ConfigBase): udp: VeilidConfigUDP tcp: VeilidConfigTCP ws: VeilidConfigWS wss: VeilidConfigWSS - def __init__(self, udp: VeilidConfigUDP, tcp: VeilidConfigTCP, ws: VeilidConfigWS, wss: VeilidConfigWSS): - self.udp = udp - self.tcp = tcp - self.ws = ws - self.wss = wss - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigProtocol( - VeilidConfigUDP.from_json(j['udp']), - VeilidConfigTCP.from_json(j['tcp']), - VeilidConfigWS.from_json(j['ws']), - VeilidConfigWSS.from_json(j['wss'])) - def to_json(self) -> dict: - return self.__dict__ - - -class VeilidConfigNetwork: +@dataclass +class VeilidConfigNetwork(ConfigBase): connection_initial_timeout_ms: int connection_inactivity_timeout_ms: int max_connections_per_ip4: int @@ -466,58 +213,9 @@ class VeilidConfigNetwork: application: VeilidConfigApplication protocol: VeilidConfigProtocol - def __init__(self, connection_initial_timeout_ms: int, connection_inactivity_timeout_ms: int, - max_connections_per_ip4: int, max_connections_per_ip6_prefix: int, - max_connections_per_ip6_prefix_size: int, max_connection_frequency_per_min: int, - client_whitelist_timeout_ms: int, reverse_connection_receipt_time_ms: int, - hole_punch_receipt_time_ms: int, routing_table: VeilidConfigRoutingTable, - rpc: VeilidConfigRPC, dht: VeilidConfigDHT, upnp: bool, detect_address_changes: bool, - restricted_nat_retries: int, tls: VeilidConfigTLS, application: VeilidConfigApplication, protocol: VeilidConfigProtocol): - self.connection_initial_timeout_ms = connection_initial_timeout_ms - self.connection_inactivity_timeout_ms = connection_inactivity_timeout_ms - self.max_connections_per_ip4 = max_connections_per_ip4 - self.max_connections_per_ip6_prefix = max_connections_per_ip6_prefix - self.max_connections_per_ip6_prefix_size = max_connections_per_ip6_prefix_size - self.max_connection_frequency_per_min = max_connection_frequency_per_min - self.client_whitelist_timeout_ms = client_whitelist_timeout_ms - self.reverse_connection_receipt_time_ms = reverse_connection_receipt_time_ms - self.hole_punch_receipt_time_ms = hole_punch_receipt_time_ms - self.routing_table = routing_table - self.rpc = rpc - self.dht = dht - self.upnp = upnp - self.detect_address_changes = detect_address_changes - self.restricted_nat_retries = restricted_nat_retries - self.tls = tls - self.application = application - self.protocol = protocol - - @staticmethod - def from_json(j: dict) -> Self: - return VeilidConfigNetwork( - j['connection_initial_timeout_ms'], - j['connection_inactivity_timeout_ms'], - j['max_connections_per_ip4'], - j['max_connections_per_ip6_prefix'], - j['max_connections_per_ip6_prefix_size'], - j['max_connection_frequency_per_min'], - j['client_whitelist_timeout_ms'], - j['reverse_connection_receipt_time_ms'], - j['hole_punch_receipt_time_ms'], - VeilidConfigRoutingTable.from_json(j['routing_table']), - VeilidConfigRPC.from_json(j['rpc']), - VeilidConfigDHT.from_json(j['dht']), - j['upnp'], - j['detect_address_changes'], - j['restricted_nat_retries'], - VeilidConfigTLS.from_json(j['tls']), - VeilidConfigApplication.from_json(j['application']), - VeilidConfigProtocol.from_json(j['protocol'])) - def to_json(self) -> dict: - return self.__dict__ - -class VeilidConfig: +@dataclass +class VeilidConfig(ConfigBase): program_name: str namespace: str capabilities: VeilidConfigCapabilities @@ -525,29 +223,3 @@ class VeilidConfig: table_store: VeilidConfigTableStore block_store: VeilidConfigBlockStore network: VeilidConfigNetwork - - def __init__(self, program_name: str, namespace: str, capabilities: VeilidConfigCapabilities, - protected_store: VeilidConfigProtectedStore, table_store: VeilidConfigTableStore, - block_store: VeilidConfigBlockStore, network: VeilidConfigNetwork): - - self.program_name = program_name - self.namespace = namespace - self.capabilities = capabilities - self.protected_store = protected_store - self.table_store = table_store - self.block_store = block_store - self.network = network - - @staticmethod - def from_json(j: dict) -> Self: - '''JSON object hook''' - return VeilidConfig(j['program_name'], j['namespace'], - VeilidConfigCapabilities.from_json(j['capabilities']), - VeilidConfigProtectedStore.from_json(j['protected_store']), - VeilidConfigTableStore.from_json(j['table_store']), - VeilidConfigBlockStore.from_json(j['block_store']), - VeilidConfigNetwork.from_json(j['network'])) - def to_json(self) -> dict: - return self.__dict__ - - \ No newline at end of file