yadunut/clan-core
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

API: type all services with dict[str,Any] in python to reduce complexity.

Browse Source
This commit is contained in:
Johannes Kirschbauer
2024-09-12 16:16:34 +02:00
parent 39518d302b
commit f2a2b8e893
8 changed files with 118 additions and 320 deletions
Download Patch File Download Diff File Expand all files Collapse all files

91
pkgs/clan-cli/clan_cli/api/admin.py
View File

@@ -1,62 +1,37 @@
from clan_cli.errors import ClanError
from clan_cli.inventory import (
AdminConfig,
ServiceAdmin,
ServiceAdminRole,
ServiceAdminRoleDefault,
ServiceMeta,
load_inventory_eval,
save_inventory,
)
# @API.register
# def set_admin_service(
# base_url: str,
# allowed_keys: dict[str, str],
# instance_name: str = "admin",
# extra_machines: list[str] | None = None,
# ) -> None:
# """
# Set the admin service of a clan
# Every machine is by default part of the admin service via the 'all' tag
# """
# if extra_machines is None:
# extra_machines = []
# inventory = load_inventory_eval(base_url)
from . import API
# if not allowed_keys:
# msg = "At least one key must be provided to ensure access"
# raise ClanError(msg)
# instance = ServiceAdmin(
# meta=ServiceMeta(name=instance_name),
# roles=ServiceAdminRole(
# default=ServiceAdminRoleDefault(
# machines=extra_machines,
# tags=["all"],
# )
# ),
# config=AdminConfig(allowedKeys=allowed_keys),
# )
@API.register
def get_admin_service(base_url: str) -> ServiceAdmin | None:
"""
Return the admin service of a clan.
# inventory.services.admin[instance_name] = instance
There is only one admin service. This might be changed in the future
"""
inventory = load_inventory_eval(base_url)
return inventory.services.admin.get("admin")
@API.register
def set_admin_service(
base_url: str,
allowed_keys: dict[str, str],
instance_name: str = "admin",
extra_machines: list[str] | None = None,
) -> None:
"""
Set the admin service of a clan
Every machine is by default part of the admin service via the 'all' tag
"""
if extra_machines is None:
extra_machines = []
inventory = load_inventory_eval(base_url)
if not allowed_keys:
msg = "At least one key must be provided to ensure access"
raise ClanError(msg)
instance = ServiceAdmin(
meta=ServiceMeta(name=instance_name),
roles=ServiceAdminRole(
default=ServiceAdminRoleDefault(
machines=extra_machines,
tags=["all"],
)
),
config=AdminConfig(allowedKeys=allowed_keys),
)
inventory.services.admin[instance_name] = instance
save_inventory(
inventory,
base_url,
f"Set admin service: '{instance_name}'",
)
# save_inventory(
# inventory,
# base_url,
# f"Set admin service: '{instance_name}'",
# )

85
pkgs/clan-cli/clan_cli/api/disk.py
View File

@@ -1,67 +1,34 @@
from clan_cli.inventory import (
ServiceMeta,
ServiceSingleDisk,
ServiceSingleDiskRole,
ServiceSingleDiskRoleDefault,
SingleDiskConfig,
load_inventory_eval,
load_inventory_json,
save_inventory,
)
from . import API
def get_instance_name(machine_name: str) -> str:
return f"{machine_name}-single-disk"
@API.register
def set_single_disk_uuid(
base_path: str,
machine_name: str,
disk_uuid: str,
) -> None:
"""
Set the disk UUID of single disk machine
"""
inventory = load_inventory_json(base_path)
# @API.register
# def set_single_disk_uuid(
# base_path: str,
# machine_name: str,
# disk_uuid: str,
# ) -> None:
# """
# Set the disk UUID of single disk machine
# """
# inventory = load_inventory_json(base_path)
instance_name = get_instance_name(machine_name)
# instance_name = get_instance_name(machine_name)
single_disk_config: ServiceSingleDisk = ServiceSingleDisk(
meta=ServiceMeta(name=instance_name),
roles=ServiceSingleDiskRole(
default=ServiceSingleDiskRoleDefault(
config=SingleDiskConfig(device=f"/dev/disk/by-id/{disk_uuid}"),
machines=[machine_name],
)
),
)
# single_disk_config: ServiceSingleDisk = ServiceSingleDisk(
# meta=ServiceMeta(name=instance_name),
# roles=ServiceSingleDiskRole(
# default=ServiceSingleDiskRoleDefault(
# config=SingleDiskConfig(device=f"/dev/disk/by-id/{disk_uuid}"),
# machines=[machine_name],
# )
# ),
# )
inventory.services.single_disk[instance_name] = single_disk_config
# inventory.services.single_disk[instance_name] = single_disk_config
save_inventory(
inventory,
base_path,
f"Set disk UUID: '{disk_uuid}' on machine: '{machine_name}'",
)
@API.register
def get_single_disk_uuid(
base_path: str,
machine_name: str,
) -> str | None:
"""
Get the disk UUID of single disk machine
"""
inventory = load_inventory_eval(base_path)
instance_name = get_instance_name(machine_name)
single_disk_config: ServiceSingleDisk = inventory.services.single_disk[
instance_name
]
return single_disk_config.roles.default.config.device
# save_inventory(
# inventory,
# base_path,
# f"Set disk UUID: '{disk_uuid}' on machine: '{machine_name}'",
# )

136
pkgs/clan-cli/clan_cli/api/iwd.py
View File

@@ -1,109 +1,67 @@
from dataclasses import dataclass
from pathlib import Path
from clan_cli.clan_uri import FlakeId
from clan_cli.errors import ClanError
from clan_cli.facts.generate import generate_facts
from clan_cli.inventory import (
IwdConfig,
IwdConfigNetwork,
ServiceIwd,
ServiceIwdRole,
ServiceIwdRoleDefault,
ServiceMeta,
load_inventory_eval,
save_inventory,
)
from clan_cli.machines.machines import Machine
from clan_cli.secrets.sops import (
maybe_get_public_key,
maybe_get_user_or_machine,
)
from . import API
def instance_name(machine_name: str) -> str:
return f"{machine_name}_wifi_0_"
@API.register
def get_iwd_service(base_url: str, machine_name: str) -> ServiceIwd:
"""
Return the admin service of a clan.
There is only one admin service. This might be changed in the future
"""
inventory = load_inventory_eval(base_url)
service_config = inventory.services.iwd.get(instance_name(machine_name))
if service_config:
return service_config
# Empty service
return ServiceIwd(
meta=ServiceMeta(name="wifi_0"),
roles=ServiceIwdRole(default=ServiceIwdRoleDefault(machines=[machine_name])),
config=IwdConfig(networks={}),
)
@dataclass
class NetworkConfig:
ssid: str
password: str
@API.register
def set_iwd_service_for_machine(
base_url: str, machine_name: str, networks: dict[str, NetworkConfig]
) -> None:
"""
Set the admin service of a clan
Every machine is by default part of the admin service via the 'all' tag
"""
_instance_name = instance_name(machine_name)
# @API.register
# def set_iwd_service_for_machine(
# base_url: str, machine_name: str, networks: dict[str, NetworkConfig]
# ) -> None:
# """
# Set the admin service of a clan
# Every machine is by default part of the admin service via the 'all' tag
# """
# _instance_name = instance_name(machine_name)
inventory = load_inventory_eval(base_url)
# inventory = load_inventory_eval(base_url)
instance = ServiceIwd(
meta=ServiceMeta(name="wifi_0"),
roles=ServiceIwdRole(
default=ServiceIwdRoleDefault(
machines=[machine_name],
)
),
config=IwdConfig(
networks={k: IwdConfigNetwork(v.ssid) for k, v in networks.items()}
),
)
# instance = ServiceIwd(
# meta=ServiceMeta(name="wifi_0"),
# roles=ServiceIwdRole(
# default=ServiceIwdRoleDefault(
# machines=[machine_name],
# )
# ),
# config=IwdConfig(
# networks={k: IwdConfigNetwork(v.ssid) for k, v in networks.items()}
# ),
# )
inventory.services.iwd[_instance_name] = instance
# inventory.services.iwd[_instance_name] = instance
save_inventory(
inventory,
base_url,
f"Set iwd service: '{_instance_name}'",
)
# save_inventory(
# inventory,
# base_url,
# f"Set iwd service: '{_instance_name}'",
# )
pubkey = maybe_get_public_key()
if not pubkey:
# TODO: do this automatically
# pubkey = generate_key()
raise ClanError(msg="No public key found. Please initialize your key.")
# pubkey = maybe_get_public_key()
# if not pubkey:
# # TODO: do this automatically
# # pubkey = generate_key()
# raise ClanError(msg="No public key found. Please initialize your key.")
registered_key = maybe_get_user_or_machine(Path(base_url), pubkey)
if not registered_key:
# TODO: do this automatically
# username = os.getlogin()
# add_user(Path(base_url), username, pubkey, force=False)
raise ClanError(msg="Your public key is not registered for use with this clan.")
# registered_key = maybe_get_user_or_machine(Path(base_url), pubkey)
# if not registered_key:
# # TODO: do this automatically
# # username = os.getlogin()
# # add_user(Path(base_url), username, pubkey, force=False)
# raise ClanError(msg="Your public key is not registered for use with this clan.")
password_dict = {f"iwd.{net.ssid}": net.password for net in networks.values()}
for net in networks.values():
generate_facts(
service=f"iwd.{net.ssid}",
machines=[Machine(machine_name, FlakeId(base_url))],
regenerate=True,
# Just returns the password
prompt=lambda service, _msg: password_dict[service],
)
# password_dict = {f"iwd.{net.ssid}": net.password for net in networks.values()}
# for net in networks.values():
# generate_facts(
# service=f"iwd.{net.ssid}",
# machines=[Machine(machine_name, FlakeId(base_url))],
# regenerate=True,
# # Just returns the password
# prompt=lambda service, _msg: password_dict[service],
# )

43
pkgs/clan-cli/clan_cli/inventory/__init__.py
View File

@@ -23,35 +23,13 @@ from clan_cli.git import commit_file
from clan_cli.nix import nix_eval
from .classes import (
AdminConfig,
Inventory,
IwdConfig,
IwdConfigNetwork,
# Machine classes
Machine,
MachineDeploy,
# General classes
Meta,
Service,
# Admin service
ServiceAdmin,
ServiceAdminRole,
ServiceAdminRoleDefault,
# Borgbackup service
ServiceBorgbackup,
ServiceBorgbackupRole,
ServiceBorgbackupRoleClient,
ServiceBorgbackupRoleServer,
# IWD
ServiceIwd,
ServiceIwdRole,
ServiceIwdRoleDefault,
ServiceMeta,
# Single Disk service
ServiceSingleDisk,
ServiceSingleDiskRole,
ServiceSingleDiskRoleDefault,
SingleDiskConfig,
)
# Re export classes here
@@ -64,27 +42,6 @@ __all__ = [
"Meta",
"Inventory",
"MachineDeploy",
"ServiceBorgbackup",
"ServiceMeta",
"ServiceBorgbackupRole",
"ServiceBorgbackupRoleClient",
"ServiceBorgbackupRoleServer",
# Single Disk service
"ServiceSingleDisk",
"ServiceSingleDiskRole",
"ServiceSingleDiskRoleDefault",
"SingleDiskConfig",
# Admin service
"ServiceAdmin",
"ServiceAdminRole",
"ServiceAdminRoleDefault",
"AdminConfig",
# IWD service,
"ServiceIwd",
"ServiceIwdRole",
"ServiceIwdRoleDefault",
"IwdConfig",
"IwdConfigNetwork",
]

5
pkgs/clan-cli/clan_cli/inventory/classes.py
View File

@@ -30,10 +30,7 @@ class Meta:
icon: None | str = field(default = None)
@dataclass
class Service:
pass
Service = dict[str, Any]
@dataclass
class Inventory:

49
pkgs/clan-cli/tests/test_deserializers.py
View File

@@ -7,18 +7,6 @@ import pytest
# Functions to test
from clan_cli.api import dataclass_to_dict, from_dict
from clan_cli.errors import ClanError
from clan_cli.inventory import (
Inventory,
Machine,
MachineDeploy,
Meta,
Service,
ServiceBorgbackup,
ServiceBorgbackupRole,
ServiceBorgbackupRoleClient,
ServiceBorgbackupRoleServer,
ServiceMeta,
)
from clan_cli.machines import machines
@@ -172,43 +160,6 @@ def test_list() -> None:
assert result == [Name("John"), Name("Sarah")]
def test_deserialize_extensive_inventory() -> None:
# TODO: Make this an abstract test, so it doesn't break the test if the inventory changes
data = {
"meta": {"name": "superclan", "description": "nice clan"},
"services": {
"borgbackup": {
"instance1": {
"meta": {
"name": "borg1",
},
"roles": {
"client": {},
"server": {},
},
}
},
},
"machines": {"foo": {"name": "foo", "deploy": {}}},
}
expected = Inventory(
meta=Meta(name="superclan", description="nice clan"),
services=Service(
borgbackup={
"instance1": ServiceBorgbackup(
meta=ServiceMeta(name="borg1"),
roles=ServiceBorgbackupRole(
client=ServiceBorgbackupRoleClient(),
server=ServiceBorgbackupRoleServer(),
),
)
}
),
machines={"foo": Machine(deploy=MachineDeploy(), name="foo")},
)
assert from_dict(Inventory, data) == expected
def test_alias_field() -> None:
@dataclass
class Person:

27
pkgs/clan-cli/tests/test_modules.py
View File

@@ -7,11 +7,6 @@ from clan_cli.clan_uri import FlakeId
from clan_cli.inventory import (
Machine,
MachineDeploy,
ServiceBorgbackup,
ServiceBorgbackupRole,
ServiceBorgbackupRoleClient,
ServiceBorgbackupRoleServer,
ServiceMeta,
load_inventory_json,
save_inventory,
)
@@ -67,18 +62,16 @@ def test_add_module_to_inventory(
inventory = load_inventory_json(base_path)
inventory.services.borgbackup = {
"borg1": ServiceBorgbackup(
meta=ServiceMeta(name="borg1"),
roles=ServiceBorgbackupRole(
client=ServiceBorgbackupRoleClient(
machines=["machine1"],
),
server=ServiceBorgbackupRoleServer(
machines=["machine1"],
),
),
)
inventory.services = {
"borgbackup": {
"borg1": {
"meta": {"name": "borg1"},
"roles": {
"client": {"machines": ["machine1"]},
"server": {"machines": ["machine1"]},
},
}
}
}
save_inventory(inventory, base_path, "Add borgbackup service")

2
pkgs/classgen/main.py
View File

@@ -201,7 +201,7 @@ def generate_dataclass(schema: dict[str, Any], class_name: str = root_class) ->
nested_classes: list[str] = []
if stop_at and class_name == stop_at:
# Skip generating classes below the stop_at property
return f"@dataclass\nclass {class_name}:\n pass\n"
return f"{class_name} = dict[str, Any]"
for prop, prop_info in properties.items():
field_name = prop.replace("-", "_")