CreatBot/CreatBotMoonraker
7
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

update_manager: use sysdeps_parser

Browse Source 
Use the sysdeps_parser utility to parse the correct dependencies
out of the data returned by the configured system dependencies
json file.

Signed-off-by:  Eric Callahan <arksine.code@gmail.com>
This commit is contained in:
Eric Callahan 2025-02-20 20:00:48 -05:00
parent 4b4ca76558
commit 31cb1fc94c
2 changed files with 13 additions and 84 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

72
moonraker/components/update_manager/app_deploy.py
View File

@ -10,13 +10,13 @@ import pathlib
import hashlib
import logging
import re
import distro
import asyncio
import importlib
from .common import AppType, Channel
from .base_deploy import BaseDeploy
from ...utils import pip_utils
from ...utils import json_wrapper as jsonw
from ...utils.sysdeps_parser import SysDepsParser
# Annotation imports
from typing import (
@ -34,9 +34,6 @@ if TYPE_CHECKING:
from ..machine import Machine
from ..file_manager.file_manager import FileManager
DISTRO_ALIASES = [distro.id()]
DISTRO_ALIASES.extend(distro.like().split())
class AppDeploy(BaseDeploy):
def __init__(self, config: ConfigHelper, prefix: str) -> None:
super().__init__(config, prefix=prefix)
@ -263,51 +260,6 @@ class AppDeploy(BaseDeploy):
svc = kconn.unit_name
await machine.do_service_action("restart", svc)
def _convert_version(self, version: str) -> Tuple[str | int, ...]:
version = version.strip()
ver_match = re.match(r"\d+(\.\d+)*((?:-|\.).+)?", version)
if ver_match is not None:
return tuple([
int(part) if part.isdigit() else part
for part in re.split(r"\.|-", version)
])
return (version,)
def _parse_system_dep(self, full_spec: str) -> str | None:
parts = full_spec.split(";", maxsplit=1)
if len(parts) == 1:
return full_spec
dep_parts = re.split(r"(==|!=|<=|>=|<|>)", parts[1].strip())
if len(dep_parts) != 3 or dep_parts[0].strip().lower() != "distro_version":
logging.info(f"Invalid requirement specifier: {full_spec}")
return None
pkg_name = parts[0].strip()
operator = dep_parts[1].strip()
distro_ver = self._convert_version(distro.version())
req_version = self._convert_version(dep_parts[2].strip())
try:
if operator == "<":
if distro_ver < req_version:
return pkg_name
elif operator == ">":
if distro_ver > req_version:
return pkg_name
elif operator == "==":
if distro_ver == req_version:
return pkg_name
elif operator == "!=":
if distro_ver != req_version:
return pkg_name
elif operator == ">=":
if distro_ver >= req_version:
return pkg_name
elif operator == "<=":
if distro_ver <= req_version:
return pkg_name
except TypeError:
pass
return None
async def _read_system_dependencies(self) -> List[str]:
eventloop = self.server.get_event_loop()
if self.system_deps_json is not None:
@ -320,26 +272,8 @@ class AppDeploy(BaseDeploy):
except Exception:
logging.exception(f"Error reading system deps: {deps_json}")
return []
for distro_id in DISTRO_ALIASES:
if distro_id in dep_info:
if not dep_info[distro_id]:
self.log_info(
f"Dependency file '{deps_json.name}' contains an empty "
f"package definition for linux distro '{distro_id}'"
)
continue
processed_deps: List[str] = []
for dep in dep_info[distro_id]:
parsed_dep = self._parse_system_dep(dep)
if parsed_dep is not None:
processed_deps.append(parsed_dep)
return processed_deps
else:
self.log_info(
f"Dependency file '{deps_json.name}' has no package definition "
f" for linux distro '{DISTRO_ALIASES[0]}'"
)
return []
parser = SysDepsParser()
return parser.parse_dependencies(dep_info)
# Fall back on install script if configured
if self.install_script is None:
return []

25
moonraker/components/update_manager/python_deploy.py
View File

@ -11,8 +11,9 @@ import logging
from enum import Enum
from ...utils.source_info import normalize_project_name, load_distribution_info
from ...utils.versions import PyVersion, GitVersion
from ...utils.sysdeps_parser import SysDepsParser
from ...utils import pip_utils, json_wrapper
from .app_deploy import AppDeploy, Channel, DISTRO_ALIASES
from .app_deploy import AppDeploy, Channel
# Annotation imports
from typing import (
@ -208,20 +209,8 @@ class PythonDeploy(AppDeploy):
if rinfo is None:
return []
dep_info = rinfo.get("system_dependencies", {})
for distro_id in DISTRO_ALIASES:
if distro_id in dep_info:
if not dep_info[distro_id]:
self.log_info(
f"Package release_info contains an empty system "
f"package definition for linux distro '{distro_id}'"
)
return dep_info[distro_id]
else:
self.log_info(
"Package release_info has no package definition "
f" for linux distro '{DISTRO_ALIASES[0]}'"
)
return []
parser = SysDepsParser()
return parser.parse_dependencies(dep_info)
async def _update_local_state(self) -> None:
self.warnings.clear()
@ -413,6 +402,12 @@ class PythonDeploy(AppDeploy):
async def _update_sys_deps(self, prev_deps: List[str]) -> None:
new_deps = self.system_deps
deps_diff = list(set(new_deps) - set(prev_deps))
if new_deps or prev_deps:
self.log_debug(
f"Pre-update system dependencies: {prev_deps}\n"
f"Post-update system dependencies: {new_deps}\n"
f"Difference to be installed: {deps_diff}"
)
if deps_diff:
await self._install_packages(deps_diff)