liz/veilid
1
0
Fork 0
Code Pull Requests Packages Releases Activity

Use dataclasses for configuration.

Browse Source
This commit is contained in:
Teknique
2023-06-16 18:17:17 -07:00
parent c278ddce53
commit 0a9cfb3338
1 changed files with 72 additions and 400 deletions
Download Patch File Download Diff File Expand all files Collapse all files

472
veilid-python/veilid/config.py
View File

@@ -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,186 +117,55 @@ 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
@@ -375,27 +173,9 @@ class VeilidConfigWS:
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
@@ -403,50 +183,17 @@ class VeilidConfigWSS:
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__