Refactor code to reduce circular dep (#4175)

* Refactor to reduce circular dep

* Update

* Update
This commit is contained in:
Daniel Girtler 2026-01-28 23:43:57 +11:00 committed by GitHub
parent 8f104bc829
commit 5612325dc3
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
17 changed files with 146 additions and 154 deletions

View File

@ -5,9 +5,8 @@ from collections.abc import Iterator
from types import TracebackType
from typing import TYPE_CHECKING, ClassVar, Self
from .command import SysCommand, SysCommandWorker
from .command import SysCommand, SysCommandWorker, locate_binary
from .exceptions import SysCallError
from .general import locate_binary
from .output import error
if TYPE_CHECKING:

View File

@ -6,12 +6,13 @@ import sys
import time
from collections.abc import Iterator
from select import EPOLLHUP, EPOLLIN, epoll
from shutil import which
from types import TracebackType
from typing import Any, Self, override
from archinstall.lib.exceptions import SysCallError
from archinstall.lib.general import clear_vt100_escape_codes, locate_binary
from archinstall.lib.exceptions import RequirementError, SysCallError
from archinstall.lib.output import debug, error, logger
from archinstall.lib.utils.encoding import clear_vt100_escape_codes
class SysCommandWorker:
@ -345,6 +346,12 @@ def run(
)
def locate_binary(name: str) -> str:
if path := which(name):
return path
raise RequirementError(f'Binary {name} does not exist.')
def _pid_exists(pid: int) -> bool:
try:
return any(subprocess.check_output(['ps', '--no-headers', '-o', 'pid', '-p', str(pid)]).strip())

View File

@ -3,10 +3,10 @@ from pathlib import Path
from typing import ClassVar
from archinstall.lib.models.device import Fido2Device
from archinstall.lib.utils.encoding import clear_vt100_escape_codes_from_str
from ..command import SysCommand, SysCommandWorker
from ..exceptions import SysCallError
from ..general import clear_vt100_escape_codes_from_str
from ..models.users import Password
from ..output import error, info

View File

@ -1,18 +1,29 @@
import json
import re
import secrets
import string
from datetime import date, datetime
from enum import Enum
from functools import lru_cache
from pathlib import Path
from shutil import which
from typing import Any, override
from archinstall.lib.exceptions import RequirementError
from archinstall.lib.packages.packages import check_package_upgrade
# https://stackoverflow.com/a/43627833/929999
_VT100_ESCAPE_REGEX = r'\x1B\[[?0-9;]*[a-zA-Z]'
_VT100_ESCAPE_REGEX_BYTES = _VT100_ESCAPE_REGEX.encode()
from .output import debug
@lru_cache(maxsize=128)
def check_version_upgrade() -> str | None:
debug('Checking version')
upgrade = None
upgrade = check_package_upgrade('archinstall')
if upgrade is None:
debug('No archinstall upgrades found')
return None
debug(f'Archinstall latest: {upgrade}')
return upgrade
def running_from_host() -> bool:
@ -26,25 +37,6 @@ def running_from_host() -> bool:
return is_host
def generate_password(length: int = 64) -> str:
haystack = string.printable # digits, ascii_letters, punctuation (!"#$[] etc) and whitespace
return ''.join(secrets.choice(haystack) for _ in range(length))
def locate_binary(name: str) -> str:
if path := which(name):
return path
raise RequirementError(f'Binary {name} does not exist.')
def clear_vt100_escape_codes(data: bytes) -> bytes:
return re.sub(_VT100_ESCAPE_REGEX_BYTES, b'', data)
def clear_vt100_escape_codes_from_str(data: str) -> str:
return re.sub(_VT100_ESCAPE_REGEX, '', data)
def jsonify(obj: Any, safe: bool = True) -> Any:
"""
Converts objects into json.dumps() compatible nested dictionaries.

View File

@ -6,7 +6,7 @@ from archinstall.lib.models.application import ApplicationConfiguration, ZramCon
from archinstall.lib.models.authentication import AuthenticationConfiguration
from archinstall.lib.models.device import DiskLayoutConfiguration, DiskLayoutType, FilesystemType, PartitionModification
from archinstall.lib.network.network_menu import ask_to_configure_network
from archinstall.lib.packages import list_available_packages
from archinstall.lib.packages.packages import ask_additional_packages_to_install, list_available_packages
from archinstall.tui.ui.menu_item import MenuItem, MenuItemGroup
from .applications.application_menu import ApplicationMenu
@ -17,7 +17,6 @@ from .configuration import save_config
from .hardware import SysInfo
from .interactions.general_conf import (
add_number_of_parallel_downloads,
ask_additional_packages_to_install,
ask_for_a_timezone,
ask_hostname,
ask_ntp,

View File

@ -32,7 +32,7 @@ from archinstall.lib.models.device import (
Unit,
)
from archinstall.lib.models.packages import Repository
from archinstall.lib.packages import installed_package
from archinstall.lib.packages.packages import installed_package
from archinstall.lib.translationhandler import tr
from .args import arch_config_handler

View File

@ -8,7 +8,6 @@ from .disk_conf import (
)
from .general_conf import (
add_number_of_parallel_downloads,
ask_additional_packages_to_install,
ask_for_a_timezone,
ask_hostname,
ask_ntp,
@ -18,7 +17,6 @@ from .system_conf import ask_for_swap, select_driver, select_kernel
__all__ = [
'add_number_of_parallel_downloads',
'ask_additional_packages_to_install',
'ask_for_a_timezone',
'ask_for_swap',
'ask_hostname',

View File

@ -2,16 +2,13 @@ import sys
from enum import Enum
from pathlib import Path
from archinstall.lib.menu.helpers import Confirmation, Input, Loading, Notify, Selection
from archinstall.lib.models.packages import Repository
from archinstall.lib.packages.packages import list_available_packages
from archinstall.lib.menu.helpers import Confirmation, Input, Selection
from archinstall.lib.translationhandler import tr
from archinstall.tui.ui.menu_item import MenuItem, MenuItemGroup
from archinstall.tui.ui.result import ResultType
from ..locale.utils import list_timezones
from ..models.packages import AvailablePackage, PackageGroup
from ..output import debug, warn
from ..output import warn
from ..translationhandler import Language
@ -132,87 +129,6 @@ def select_archinstall_language(languages: list[Language], preset: Language) ->
raise ValueError('Language selection not handled')
def ask_additional_packages_to_install(
preset: list[str] = [],
repositories: set[Repository] = set(),
) -> list[str]:
repositories |= {Repository.Core, Repository.Extra}
respos_text = ', '.join(r.value for r in repositories)
output = tr('Repositories: {}').format(respos_text) + '\n'
output += tr('Loading packages...')
result = Loading[dict[str, AvailablePackage]](
header=output,
data_callback=lambda: list_available_packages(tuple(repositories)),
).show()
if result.type_ != ResultType.Selection:
debug('Error while loading packages')
return preset
packages = result.get_value()
if not packages:
Notify(tr('No packages found')).show()
return []
package_groups = PackageGroup.from_available_packages(packages)
# Additional packages (with some light weight error handling for invalid package names)
header = tr('Only packages such as base, sudo, linux, linux-firmware, efibootmgr and optional profile packages are installed.') + '\n'
header += tr('Note: base-devel is no longer installed by default. Add it here if you need build tools.') + '\n'
header += tr('Select any packages from the below list that should be installed additionally') + '\n'
# there are over 15k packages so this needs to be quick
preset_packages: list[AvailablePackage | PackageGroup] = []
for p in preset:
if p in packages:
preset_packages.append(packages[p])
elif p in package_groups:
preset_packages.append(package_groups[p])
items = [
MenuItem(
name,
value=pkg,
preview_action=lambda x: x.value.info() if x.value else None,
)
for name, pkg in packages.items()
]
items += [
MenuItem(
name,
value=group,
preview_action=lambda x: x.value.info() if x.value else None,
)
for name, group in package_groups.items()
]
menu_group = MenuItemGroup(items, sort_items=True)
menu_group.set_selected_by_value(preset_packages)
pck_result = Selection[AvailablePackage | PackageGroup](
menu_group,
header=header,
allow_reset=True,
allow_skip=True,
multi=True,
preview_location='right',
enable_filter=True,
).show()
match pck_result.type_:
case ResultType.Skip:
return preset
case ResultType.Reset:
return []
case ResultType.Selection:
selected_pacakges = pck_result.get_values()
return [pkg.name for pkg in selected_pacakges]
def add_number_of_parallel_downloads(preset: int = 1) -> int | None:
max_recommended = 5

View File

@ -6,10 +6,10 @@ from types import TracebackType
from archinstall.lib.disk.utils import get_lsblk_info, umount
from archinstall.lib.models.device import DEFAULT_ITER_TIME
from archinstall.lib.utils.util import generate_password
from .command import SysCommand, SysCommandWorker, run
from .exceptions import DiskError, SysCallError
from .general import generate_password
from .models.users import Password
from .output import debug, info

View File

@ -8,7 +8,7 @@ from enum import Enum
from pathlib import Path
from typing import TYPE_CHECKING, Any
from .utils.unicode import unicode_ljust, unicode_rjust
from .utils.encoding import unicode_ljust, unicode_rjust
if TYPE_CHECKING:
from _typeshed import DataclassInstance

View File

@ -1,6 +0,0 @@
from .packages import installed_package, list_available_packages
__all__ = [
'installed_package',
'list_available_packages',
]

View File

@ -1,29 +1,16 @@
from functools import lru_cache
from archinstall.lib.menu.helpers import Loading, Notify, Selection
from archinstall.lib.models.packages import AvailablePackage, LocalPackage, PackageGroup, Repository
from archinstall.lib.translationhandler import tr
from archinstall.tui.ui.menu_item import MenuItem, MenuItemGroup
from archinstall.tui.ui.result import ResultType
from ..exceptions import SysCallError
from ..models.packages import AvailablePackage, LocalPackage, Repository
from ..output import debug
from ..pacman.pacman import Pacman
# TODO: This shouldn't be living in here but there are too many
# circular dependecies so they will need to be cleanup up first
@lru_cache(maxsize=128)
def check_version_upgrade() -> str | None:
debug('Checking version')
upgrade = None
upgrade = check_package_upgrade('archinstall')
if upgrade is None:
debug('No archinstall upgrades found')
return None
debug(f'Archinstall latest: {upgrade}')
return upgrade
def installed_package(package: str) -> LocalPackage | None:
try:
package_info = []
@ -96,3 +83,84 @@ def _parse_package_output[PackageType: (AvailablePackage, LocalPackage)](
package[key] = value.strip()
return cls.model_validate(package)
def ask_additional_packages_to_install(
preset: list[str] = [],
repositories: set[Repository] = set(),
) -> list[str]:
repositories |= {Repository.Core, Repository.Extra}
respos_text = ', '.join(r.value for r in repositories)
output = tr('Repositories: {}').format(respos_text) + '\n'
output += tr('Loading packages...')
result = Loading[dict[str, AvailablePackage]](
header=output,
data_callback=lambda: list_available_packages(tuple(repositories)),
).show()
if result.type_ != ResultType.Selection:
debug('Error while loading packages')
return preset
packages = result.get_value()
if not packages:
Notify(tr('No packages found')).show()
return []
package_groups = PackageGroup.from_available_packages(packages)
# Additional packages (with some light weight error handling for invalid package names)
header = tr('Only packages such as base, sudo, linux, linux-firmware, efibootmgr and optional profile packages are installed.') + '\n'
header += tr('Note: base-devel is no longer installed by default. Add it here if you need build tools.') + '\n'
header += tr('Select any packages from the below list that should be installed additionally') + '\n'
# there are over 15k packages so this needs to be quick
preset_packages: list[AvailablePackage | PackageGroup] = []
for p in preset:
if p in packages:
preset_packages.append(packages[p])
elif p in package_groups:
preset_packages.append(package_groups[p])
items = [
MenuItem(
name,
value=pkg,
preview_action=lambda x: x.value.info() if x.value else None,
)
for name, pkg in packages.items()
]
items += [
MenuItem(
name,
value=group,
preview_action=lambda x: x.value.info() if x.value else None,
)
for name, group in package_groups.items()
]
menu_group = MenuItemGroup(items, sort_items=True)
menu_group.set_selected_by_value(preset_packages)
pck_result = Selection[AvailablePackage | PackageGroup](
menu_group,
header=header,
allow_reset=True,
allow_skip=True,
multi=True,
preview_location='right',
enable_filter=True,
).show()
match pck_result.type_:
case ResultType.Skip:
return preset
case ResultType.Reset:
return []
case ResultType.Selection:
selected_pacakges = pck_result.get_values()
return [pkg.name for pkg in selected_pacakges]

View File

@ -1,6 +1,19 @@
import re
import unicodedata
from functools import lru_cache
# https://stackoverflow.com/a/43627833/929999
_VT100_ESCAPE_REGEX = r'\x1B\[[?0-9;]*[a-zA-Z]'
_VT100_ESCAPE_REGEX_BYTES = _VT100_ESCAPE_REGEX.encode()
def clear_vt100_escape_codes(data: bytes) -> bytes:
return re.sub(_VT100_ESCAPE_REGEX_BYTES, b'', data)
def clear_vt100_escape_codes_from_str(data: str) -> str:
return re.sub(_VT100_ESCAPE_REGEX, '', data)
@lru_cache(maxsize=128)
def _is_wide_character(char: str) -> bool:

View File

@ -1,3 +1,5 @@
import secrets
import string
from pathlib import Path
from archinstall.lib.menu.helpers import Input
@ -8,6 +10,11 @@ from ..models.users import Password
from ..output import FormattedOutput
def generate_password(length: int = 64) -> str:
haystack = string.printable # digits, ascii_letters, punctuation (!"#$[] etc) and whitespace
return ''.join(secrets.choice(haystack) for _ in range(length))
def get_password(
header: str | None = None,
allow_skip: bool = False,

View File

@ -10,10 +10,9 @@ from pathlib import Path
from archinstall.lib.args import arch_config_handler
from archinstall.lib.disk.utils import disk_layouts
from archinstall.lib.general import running_from_host
from archinstall.lib.general import check_version_upgrade, running_from_host
from archinstall.lib.network.wifi_handler import WifiHandler
from archinstall.lib.networking import ping
from archinstall.lib.packages.packages import check_version_upgrade
from .lib.hardware import SysInfo
from .lib.output import debug, error, info, warn

View File

@ -8,6 +8,7 @@ from archinstall.lib.authentication.authentication_handler import Authentication
from archinstall.lib.configuration import ConfigurationOutput
from archinstall.lib.disk.filesystem import FilesystemHandler
from archinstall.lib.disk.utils import disk_layouts
from archinstall.lib.general import check_version_upgrade
from archinstall.lib.global_menu import GlobalMenu
from archinstall.lib.hardware import SysInfo
from archinstall.lib.installer import Installer, accessibility_tools_in_use, run_custom_user_commands
@ -21,7 +22,6 @@ from archinstall.lib.models.device import (
from archinstall.lib.models.users import User
from archinstall.lib.network.network_handler import NetworkHandler
from archinstall.lib.output import debug, error, info
from archinstall.lib.packages.packages import check_version_upgrade
from archinstall.lib.profile.profiles_handler import profile_handler
from archinstall.lib.translationhandler import tr

View File

@ -6,7 +6,7 @@ from typing import Any, ClassVar, Self, override
from archinstall.lib.translationhandler import tr
from ..lib.utils.unicode import unicode_ljust
from ..lib.utils.encoding import unicode_ljust
@dataclass