removed node classes as currently may not be necessary
This commit is contained in:
152
anodes.py
152
anodes.py
@@ -1,152 +0,0 @@
|
||||
"""
|
||||
Library of classes modeling Ansible nodes or their types.
|
||||
"""
|
||||
|
||||
from enum import Enum
|
||||
from pathlib import Path, PurePath
|
||||
from typing import TypedDict as Dict
|
||||
from typing import Union, Literal, Required, Self
|
||||
from custtypes import ExecutedPath, IdlePath, VirtualPrivateServers, AnsibleScopes
|
||||
from softman import Software, SoftPathGroup, SoftScope, Apps, Softs
|
||||
from whereami import USER_PATH, PROJ_ROOT
|
||||
from ansible_vault import Vault
|
||||
import secrets
|
||||
|
||||
class ControlNode:
|
||||
__user_path: ExecutedPath = USER_PATH
|
||||
__proj_root: ExecutedPath = PROJ_ROOT
|
||||
__conf_paths: tuple[IdlePath] | list[IdlePath] = (
|
||||
PurePath("/etc"),
|
||||
PurePath("/usr", "local", "etc"),
|
||||
PurePath(str(__user_path), ".config")
|
||||
)
|
||||
__data_paths: tuple[IdlePath] | list[IdlePath] = (
|
||||
PurePath("/usr", "local", "share"),
|
||||
PurePath(str(__user_path), ".local", "share")
|
||||
)
|
||||
|
||||
def __init__(self, ansible_proj_root: ExecutedPath | None = None):
|
||||
if ansible_proj_root is not None:
|
||||
self.__proj_root = ansible_proj_root
|
||||
role_paths: tuple[IdlePath] | list[IdlePath] = (
|
||||
PurePath(str(self.__proj_root), "roles"),
|
||||
PurePath(str(self.__proj_root), ".ansible/roles")
|
||||
)
|
||||
roledata_paths: tuple[IdlePath] | list[IdlePath] = (
|
||||
PurePath(str(self.proj_roles[0]), "files"),
|
||||
PurePath(str(self.proj_roles[0]), "templates"),
|
||||
PurePath(str(self.proj_roles[1]), "files"),
|
||||
PurePath(str(self.proj_roles[1]), "templates")
|
||||
)
|
||||
setattr(self, AnsibleScopes.ROLE.name, {
|
||||
"data": roledata_paths,
|
||||
"vars": role_paths
|
||||
})
|
||||
setattr(self, AnsibleScopes.GROUPVARS.name, {
|
||||
"vars": (PurePath(str(self.__proj_root), "group_vars"),)
|
||||
})
|
||||
setattr(self, AnsibleScopes.HOSTVARS.name, {
|
||||
"vars": (PurePath(str(self.__proj_root), "host_vars"),)
|
||||
})
|
||||
setattr(self, AnsibleScopes.INVENTORY.name, {
|
||||
"vars": (PurePath(str(self.__proj_root), "hosts.yml"),)
|
||||
})
|
||||
|
||||
def get_scope(self, scope: AnsibleScopes = AnsibleScopes.INVENTORY.name, index = 0):
|
||||
return getattr(self, scope)[index]
|
||||
|
||||
@property
|
||||
def home(self) -> ExecutedPath:
|
||||
return self.__user_path
|
||||
|
||||
@property
|
||||
def root(self) -> ExecutedPath:
|
||||
return self.__proj_root
|
||||
|
||||
@property
|
||||
def sys_confs(self) -> ExecutedPath:
|
||||
return self.__conf_paths
|
||||
|
||||
@property
|
||||
def sys_data(self) -> ExecutedPath:
|
||||
return self.__data_paths
|
||||
|
||||
# userSSHParams = Dict("userSSHParams", {
|
||||
# "username": Required[str],
|
||||
# "paths": Apps,
|
||||
# "keys": dict,
|
||||
# "password": Required[str],
|
||||
# "fate": Literal["disposal", "retention"]
|
||||
# }, total=False)
|
||||
|
||||
vpsSchema = Dict("vpsSchema", {
|
||||
"fqdn": Required[str],
|
||||
"vps_service": {
|
||||
"exists": Required[bool],
|
||||
"password": Required[str],
|
||||
"api_key": Required[str],
|
||||
"type": Required[Literal["linode"]],
|
||||
"region": Literal["us-east"],
|
||||
"ssh_authorized_keys": list[IdlePath | ExecutedPath | str],
|
||||
"ssh_private_key_paths": list[IdlePath | ExecutedPath | str],
|
||||
"ssh_private_key_path_pref": int,
|
||||
"root_fate": Required[Literal["disposal","retention"]],
|
||||
"ssh_motd_script_basenames": list[str]
|
||||
},
|
||||
"keywords": list[str]
|
||||
}, total=False)
|
||||
|
||||
class RemoteNode:
|
||||
# __user_path =
|
||||
_fqdn: str | None = None
|
||||
|
||||
def __init__(self, cnode: ControlNode, api_key: str = secrets.token_urlsafe(32), password: str = "password123", name: str = ".test", service: VirtualPrivateServers = VirtualPrivateServers.Linode.name, region: Literal["us-east"] | None = "us-east", keywords: list[str] | None = ["server"]):
|
||||
self.root: dict = dict()
|
||||
self.root["username"]: str = "root"
|
||||
vault = Vault(password)
|
||||
self.root["password"]: str = vault.dump(password)
|
||||
self.root["software"]: Software = Software()
|
||||
self.owner = cnode
|
||||
|
||||
app_input = {
|
||||
SoftPathGroup.CONFIG.name: {
|
||||
SoftScope.PERSONAL.name: PurePath(str(cnode.home), ("." + Softs.ssh.name))
|
||||
},
|
||||
SoftPathGroup.DATA.name: {
|
||||
SoftScope.GLOBAL.name: PurePath(str(cnode.sys_confs[0]), "update-motd.d")
|
||||
}
|
||||
}
|
||||
self.root["software"].append(Softs.ssh.name, **app_input)
|
||||
|
||||
self._fqdn = name
|
||||
self.root["software"]._fqdn = name
|
||||
|
||||
# root_ssh_input: userSSHParams = {
|
||||
root_ssh_input = {
|
||||
"username": self.root["username"],
|
||||
"password": self.root["password"]
|
||||
}
|
||||
self.ssh: UserSSH = UserSSH(**root_ssh_input)
|
||||
|
||||
self.apps: list = self.root["software"].show(contents = True)
|
||||
self.keywords = keywords
|
||||
self._api_key: str | None = api_key
|
||||
self.service = service
|
||||
self.region = region
|
||||
self.model: dict | None = None
|
||||
|
||||
def set_region(self, name: Literal["us-east"] = "us-east") -> None:
|
||||
self.region = name
|
||||
|
||||
def set_password(self, password: str) -> None:
|
||||
vault = Vault(self.root["password"])
|
||||
self.root["password"] = vault.dump(self.root["password"])
|
||||
|
||||
def set_api(self, key: str) -> None:
|
||||
vault = Vault(self._api_key)
|
||||
self._api_key = vault.dump(self._api_key)
|
||||
|
||||
def add_tags(self, *name):
|
||||
self.keywords: list = []
|
||||
self.keywords += list(name)
|
||||
|
||||
Reference in New Issue
Block a user