jeka/kiauh
1
0
Fork 0
mirror of https://github.com/dw-0/kiauh.git synced 2025-12-25 16:53:36 +05:00
Code Issues Packages Projects Releases Wiki Activity

refactor: more robust type hinting

Browse Source 
Signed-off-by: Dominik Willner <th33xitus@gmail.com>
This commit is contained in:
dw-0
2024-05-19 18:41:29 +02:00
parent 6eb06772b4
commit ac0478b062
10 changed files with 167 additions and 235 deletions
Download Patch File Download Diff File Expand all files Collapse all files

27
kiauh/components/crowsnest/crowsnest.py
View File

@@ -12,19 +12,19 @@ import shutil
import time
from pathlib import Path
from subprocess import CalledProcessError, run
from typing import Dict, List, Literal, Union
from typing import List
from components.crowsnest import CROWSNEST_BACKUP_DIR, CROWSNEST_DIR, CROWSNEST_REPO
from components.klipper.klipper import Klipper
from core.backup_manager.backup_manager import BackupManager
from core.instance_manager.instance_manager import InstanceManager
from core.settings.kiauh_settings import KiauhSettings
from utils.common import check_install_dependencies, get_install_status
from utils.common import (
check_install_dependencies,
get_install_status,
)
from utils.constants import CURRENT_USER
from utils.git_utils import (
get_local_commit,
get_remote_commit,
get_repo_name,
git_clone_wrapper,
git_pull_wrapper,
)
@@ -34,6 +34,7 @@ from utils.sys_utils import (
cmd_sysctl_service,
parse_packages_from_file,
)
from utils.types import ComponentStatus
def install_crowsnest() -> None:
@@ -142,25 +143,13 @@ def update_crowsnest() -> None:
return
def get_crowsnest_status() -> (
Dict[
Literal["status", "status_code", "repo", "local", "remote"],
Union[str, int],
]
):
def get_crowsnest_status() -> ComponentStatus:
files = [
Path("/usr/local/bin/crowsnest"),
Path("/etc/logrotate.d/crowsnest"),
Path("/etc/systemd/system/crowsnest.service"),
]
status = get_install_status(CROWSNEST_DIR, files)
return {
"status": status.get("status"),
"status_code": status.get("status_code"),
"repo": get_repo_name(CROWSNEST_DIR),
"local": get_local_commit(CROWSNEST_DIR),
"remote": get_remote_commit(CROWSNEST_DIR),
}
return get_install_status(CROWSNEST_DIR, files=files)
def remove_crowsnest() -> None:

23
kiauh/components/klipper/klipper_utils.py
View File

@@ -12,7 +12,7 @@ import os
import re
import shutil
from subprocess import CalledProcessError, run
from typing import Dict, List, Literal, Optional, Union
from typing import Dict, List, Optional, Union
from components.klipper import (
KLIPPER_BACKUP_DIR,
@@ -39,29 +39,16 @@ from core.instance_manager.base_instance import BaseInstance
from core.instance_manager.instance_manager import InstanceManager
from core.instance_manager.name_scheme import NameScheme
from utils import PRINTER_CFG_BACKUP_DIR
from utils.common import get_install_status_common
from utils.common import get_install_status
from utils.constants import CURRENT_USER
from utils.git_utils import get_local_commit, get_remote_commit, get_repo_name
from utils.input_utils import get_confirm, get_number_input, get_string_input
from utils.logger import DialogType, Logger
from utils.sys_utils import cmd_sysctl_service
from utils.types import ComponentStatus
def get_klipper_status() -> (
Dict[
Literal["status", "status_code", "instances", "repo", "local", "remote"],
Union[str, int],
]
):
status = get_install_status_common(Klipper, KLIPPER_DIR, KLIPPER_ENV_DIR)
return {
"status": status.get("status"),
"status_code": status.get("status_code"),
"instances": status.get("instances"),
"repo": get_repo_name(KLIPPER_DIR),
"local": get_local_commit(KLIPPER_DIR),
"remote": get_remote_commit(KLIPPER_DIR),
}
def get_klipper_status() -> ComponentStatus:
return get_install_status(KLIPPER_DIR, KLIPPER_ENV_DIR, Klipper)
def check_is_multi_install(

44
kiauh/components/klipperscreen/klipperscreen.py
View File

@@ -8,39 +8,40 @@
# ======================================================================= #
import shutil
from pathlib import Path
from subprocess import run, CalledProcessError
from typing import List, Dict, Literal, Union
from subprocess import CalledProcessError, run
from typing import List
from components.klipper.klipper import Klipper
from components.klipperscreen import (
KLIPPERSCREEN_DIR,
KLIPPERSCREEN_REPO,
KLIPPERSCREEN_ENV,
KLIPPERSCREEN_BACKUP_DIR,
KLIPPERSCREEN_DIR,
KLIPPERSCREEN_ENV,
KLIPPERSCREEN_REPO,
)
from components.moonraker.moonraker import Moonraker
from core.backup_manager.backup_manager import BackupManager
from core.instance_manager.instance_manager import InstanceManager
from core.settings.kiauh_settings import KiauhSettings
from utils.common import get_install_status, check_install_dependencies
from utils.common import (
check_install_dependencies,
get_install_status,
)
from utils.config_utils import add_config_section, remove_config_section
from utils.constants import SYSTEMD
from utils.fs_utils import remove_with_sudo
from utils.git_utils import (
git_clone_wrapper,
git_pull_wrapper,
get_repo_name,
get_local_commit,
get_remote_commit,
)
from utils.input_utils import get_confirm
from utils.logger import Logger, DialogType
from utils.logger import DialogType, Logger
from utils.sys_utils import (
check_python_version,
cmd_sysctl_manage,
cmd_sysctl_service,
install_python_requirements,
cmd_sysctl_manage,
)
from utils.types import ComponentStatus
def install_klipperscreen() -> None:
@@ -141,25 +142,12 @@ def update_klipperscreen() -> None:
return
def get_klipperscreen_status() -> (
Dict[
Literal["status", "status_code", "repo", "local", "remote"],
Union[str, int],
]
):
files = [
def get_klipperscreen_status() -> ComponentStatus:
return get_install_status(
KLIPPERSCREEN_DIR,
KLIPPERSCREEN_ENV,
SYSTEMD.joinpath("KlipperScreen.service"),
]
status = get_install_status(KLIPPERSCREEN_DIR, files)
return {
"status": status.get("status"),
"status_code": status.get("status_code"),
"repo": get_repo_name(KLIPPERSCREEN_DIR),
"local": get_local_commit(KLIPPERSCREEN_DIR),
"remote": get_remote_commit(KLIPPERSCREEN_DIR),
}
files=[SYSTEMD.joinpath("KlipperScreen.service")],
)
def remove_klipperscreen() -> None:

27
kiauh/components/mobileraker/mobileraker.py
View File

@@ -9,7 +9,7 @@
import shutil
from pathlib import Path
from subprocess import CalledProcessError, run
from typing import Dict, List, Literal, Union
from typing import List
from components.klipper.klipper import Klipper
from components.mobileraker import (
@@ -27,9 +27,6 @@ from utils.config_utils import add_config_section, remove_config_section
from utils.constants import SYSTEMD
from utils.fs_utils import remove_with_sudo
from utils.git_utils import (
get_local_commit,
get_remote_commit,
get_repo_name,
git_clone_wrapper,
git_pull_wrapper,
)
@@ -41,6 +38,7 @@ from utils.sys_utils import (
cmd_sysctl_service,
install_python_requirements,
)
from utils.types import ComponentStatus
def install_mobileraker() -> None:
@@ -138,25 +136,12 @@ def update_mobileraker() -> None:
return
def get_mobileraker_status() -> (
Dict[
Literal["status", "status_code", "repo", "local", "remote"],
Union[str, int],
]
):
files = [
def get_mobileraker_status() -> ComponentStatus:
return get_install_status(
MOBILERAKER_DIR,
MOBILERAKER_ENV,
SYSTEMD.joinpath("mobileraker.service"),
]
status = get_install_status(MOBILERAKER_DIR, files)
return {
"status": status.get("status"),
"status_code": status.get("status_code"),
"repo": get_repo_name(MOBILERAKER_DIR),
"local": get_local_commit(MOBILERAKER_DIR),
"remote": get_remote_commit(MOBILERAKER_DIR),
}
files=[SYSTEMD.joinpath("mobileraker.service")],
)
def remove_mobileraker() -> None:

27
kiauh/components/moonraker/moonraker_utils.py
View File

@@ -8,15 +8,15 @@
# ======================================================================= #
import shutil
from typing import Dict, Literal, List, Union, Optional
from typing import Dict, List, Optional
from components.moonraker import (
DEFAULT_MOONRAKER_PORT,
MODULE_PATH,
MOONRAKER_DIR,
MOONRAKER_ENV_DIR,
MOONRAKER_BACKUP_DIR,
MOONRAKER_DB_BACKUP_DIR,
MOONRAKER_DIR,
MOONRAKER_ENV_DIR,
)
from components.moonraker.moonraker import Moonraker
from components.webui_client.base_data import BaseWebClient
@@ -25,29 +25,16 @@ from components.webui_client.mainsail_data import MainsailData
from core.backup_manager.backup_manager import BackupManager
from core.config_manager.config_manager import ConfigManager
from core.instance_manager.instance_manager import InstanceManager
from utils.common import get_install_status_common
from utils.git_utils import get_repo_name, get_local_commit, get_remote_commit
from utils.common import get_install_status
from utils.logger import Logger
from utils.sys_utils import (
get_ipv4_addr,
)
from utils.types import ComponentStatus
def get_moonraker_status() -> (
Dict[
Literal["status", "status_code", "instances", "repo", "local", "remote"],
Union[str, int],
]
):
status = get_install_status_common(Moonraker, MOONRAKER_DIR, MOONRAKER_ENV_DIR)
return {
"status": status.get("status"),
"status_code": status.get("status_code"),
"instances": status.get("instances"),
"repo": get_repo_name(MOONRAKER_DIR),
"local": get_local_commit(MOONRAKER_DIR),
"remote": get_remote_commit(MOONRAKER_DIR),
}
def get_moonraker_status() -> ComponentStatus:
return get_install_status(MOONRAKER_DIR, MOONRAKER_ENV_DIR, Moonraker)
def create_example_moonraker_conf(

52
kiauh/components/webui_client/client_utils.py
View File

@@ -7,61 +7,53 @@
# This file may be distributed under the terms of the GNU GPLv3 license #
# ======================================================================= #
import json
import json # noqa: I001
import shutil
from pathlib import Path
from typing import List, Dict, Literal, Union, get_args
from typing import List, get_args
from components.klipper.klipper import Klipper
from components.webui_client.base_data import (
WebClientType,
BaseWebClient,
BaseWebClientConfig,
WebClientType,
)
from components.webui_client.mainsail_data import MainsailData
from core.backup_manager.backup_manager import BackupManager
from core.settings.kiauh_settings import KiauhSettings
from utils import NGINX_SITES_AVAILABLE, NGINX_CONFD
from utils.common import get_install_status_webui
from utils.constants import COLOR_CYAN, RESET_FORMAT, COLOR_YELLOW
from utils import NGINX_CONFD, NGINX_SITES_AVAILABLE
from utils.common import get_install_status
from utils.constants import COLOR_CYAN, COLOR_YELLOW, RESET_FORMAT
from utils.git_utils import (
get_latest_tag,
get_latest_unstable_tag,
get_repo_name,
get_local_commit,
get_remote_commit,
)
from utils.logger import Logger
from utils.types import ComponentStatus, InstallStatus
def get_client_status(
client: BaseWebClient, fetch_remote: bool = False
) -> Dict[Literal["status", "local", "remote"], str]:
status = get_install_status_webui(
client.client_dir,
) -> ComponentStatus:
files = [
NGINX_SITES_AVAILABLE.joinpath(client.name),
NGINX_CONFD.joinpath("upstreams.conf"),
NGINX_CONFD.joinpath("common_vars.conf"),
)
local = get_local_client_version(client)
remote = get_remote_client_version(client) if fetch_remote else None
return {"status": status, "local": local, "remote": remote}
]
status = get_install_status(client.client_dir, files=files)
# if the client dir does not exist, set the status to not
# installed even if the other files are present
if not client.client_dir.exists():
status["status"] = InstallStatus.NOT_INSTALLED
status["local"] = get_local_client_version(client)
status["remote"] = get_remote_client_version(client) if fetch_remote else None
return status
def get_client_config_status(
client: BaseWebClient,
) -> Dict[
Literal["repo", "local", "remote"],
Union[str, int],
]:
client_config = client.client_config.config_dir
return {
"repo": get_repo_name(client_config),
"local": get_local_commit(client_config),
"remote": get_remote_commit(client_config),
}
def get_client_config_status(client: BaseWebClient) -> ComponentStatus:
return get_install_status(client.client_config.config_dir)
def get_current_client_config(clients: List[BaseWebClient]) -> str: