Remove some Any instances from the codebase (#3221)

This commit is contained in:
correctmost 2025-03-02 16:24:36 -05:00 committed by GitHub
parent d6ee5a66b2
commit 5f7b16b152
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
7 changed files with 61 additions and 29 deletions

View File

@ -8,7 +8,7 @@ import time
import traceback import traceback
from argparse import ArgumentParser, Namespace from argparse import ArgumentParser, Namespace
from pathlib import Path from pathlib import Path
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING
from archinstall.lib.args import arch_config_handler from archinstall.lib.args import arch_config_handler
from archinstall.lib.disk.utils import disk_layouts from archinstall.lib.disk.utils import disk_layouts

View File

@ -45,7 +45,9 @@ from .plugins import plugins
from .storage import storage from .storage import storage
if TYPE_CHECKING: if TYPE_CHECKING:
_: Any from archinstall.lib.translationhandler import DeferredTranslation
_: Callable[[str], DeferredTranslation]
# Any package that the Installer() is responsible for (optional and the default ones) # Any package that the Installer() is responsible for (optional and the default ones)
__packages__ = ["base", "base-devel", "linux-firmware", "linux", "linux-lts", "linux-zen", "linux-hardened"] __packages__ = ["base", "base-devel", "linux-firmware", "linux", "linux-lts", "linux-zen", "linux-hardened"]
@ -80,7 +82,7 @@ class Installer:
self.init_time = time.strftime('%Y-%m-%d_%H-%M-%S') self.init_time = time.strftime('%Y-%m-%d_%H-%M-%S')
self.milliseconds = int(str(time.time()).split('.')[1]) self.milliseconds = int(str(time.time()).split('.')[1])
self.helper_flags: dict[str, Any] = {'base': False, 'bootloader': None} self.helper_flags: dict[str, str | bool | None] = {'base': False, 'bootloader': None}
for kernel in self.kernels: for kernel in self.kernels:
self._base_packages.append(kernel) self._base_packages.append(kernel)
@ -162,23 +164,21 @@ class Installer:
""" """
if not arch_config_handler.args.skip_ntp: if not arch_config_handler.args.skip_ntp:
info(_('Waiting for time sync (timedatectl show) to complete.')) info(str(_('Waiting for time sync (timedatectl show) to complete.')))
started_wait = time.time() started_wait = time.time()
notified = False notified = False
while True: while True:
if not notified and time.time() - started_wait > 5: if not notified and time.time() - started_wait > 5:
notified = True notified = True
warn( warn(str(_("Time synchronization not completing, while you wait - check the docs for workarounds: https://archinstall.readthedocs.io/")))
_("Time synchronization not completing, while you wait - check the docs for workarounds: https://archinstall.readthedocs.io/"))
time_val = SysCommand('timedatectl show --property=NTPSynchronized --value').decode() time_val = SysCommand('timedatectl show --property=NTPSynchronized --value').decode()
if time_val and time_val.strip() == 'yes': if time_val and time_val.strip() == 'yes':
break break
time.sleep(1) time.sleep(1)
else: else:
info( info(str(_('Skipping waiting for automatic time sync (this can cause issues if time is out of sync during installation)')))
_('Skipping waiting for automatic time sync (this can cause issues if time is out of sync during installation)'))
info('Waiting for automatic mirror selection (reflector) to complete.') info('Waiting for automatic mirror selection (reflector) to complete.')
while self._service_state('reflector') not in ('dead', 'failed', 'exited'): while self._service_state('reflector') not in ('dead', 'failed', 'exited'):
@ -188,7 +188,7 @@ class Installer:
# while self._service_state('pacman-init') not in ('dead', 'failed', 'exited'): # while self._service_state('pacman-init') not in ('dead', 'failed', 'exited'):
# time.sleep(1) # time.sleep(1)
info(_('Waiting for Arch Linux keyring sync (archlinux-keyring-wkd-sync) to complete.')) info(str(_('Waiting for Arch Linux keyring sync (archlinux-keyring-wkd-sync) to complete.')))
# Wait for the timer to kick in # Wait for the timer to kick in
while self._service_started('archlinux-keyring-wkd-sync.timer') is None: while self._service_started('archlinux-keyring-wkd-sync.timer') is None:
time.sleep(1) time.sleep(1)

View File

@ -2,13 +2,14 @@ from __future__ import annotations
from dataclasses import dataclass, field from dataclasses import dataclass, field
from enum import Enum from enum import Enum
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, NotRequired, TypedDict
from ..models.profile_model import ProfileConfiguration from ..models.profile_model import ProfileConfiguration
if TYPE_CHECKING: if TYPE_CHECKING:
from collections.abc import Callable from collections.abc import Callable
from archinstall.lib.installer import Installer
from archinstall.lib.translationhandler import DeferredTranslation from archinstall.lib.translationhandler import DeferredTranslation
_: Callable[[str], DeferredTranslation] _: Callable[[str], DeferredTranslation]
@ -29,6 +30,14 @@ class NicType(Enum):
return str(_('Manual configuration')) return str(_('Manual configuration'))
class _NicSerialization(TypedDict):
iface: str | None
ip: str | None
dhcp: bool
gateway: str | None
dns: list[str]
@dataclass @dataclass
class Nic: class Nic:
iface: str | None = None iface: str | None = None
@ -37,7 +46,7 @@ class Nic:
gateway: str | None = None gateway: str | None = None
dns: list[str] = field(default_factory=list) dns: list[str] = field(default_factory=list)
def table_data(self) -> dict[str, Any]: def table_data(self) -> dict[str, str | bool | list[str]]:
return { return {
'iface': self.iface if self.iface else '', 'iface': self.iface if self.iface else '',
'ip': self.ip if self.ip else '', 'ip': self.ip if self.ip else '',
@ -46,7 +55,7 @@ class Nic:
'dns': self.dns 'dns': self.dns
} }
def json(self) -> dict[str, Any]: def json(self) -> _NicSerialization:
return { return {
'iface': self.iface, 'iface': self.iface,
'ip': self.ip, 'ip': self.ip,
@ -56,7 +65,7 @@ class Nic:
} }
@staticmethod @staticmethod
def parse_arg(arg: dict[str, Any]) -> Nic: def parse_arg(arg: _NicSerialization) -> Nic:
return Nic( return Nic(
iface=arg.get('iface', None), iface=arg.get('iface', None),
ip=arg.get('ip', None), ip=arg.get('ip', None),
@ -93,20 +102,25 @@ class Nic:
return config_str return config_str
class _NetworkConfigurationSerialization(TypedDict):
type: str
nics: NotRequired[list[_NicSerialization]]
@dataclass @dataclass
class NetworkConfiguration: class NetworkConfiguration:
type: NicType type: NicType
nics: list[Nic] = field(default_factory=list) nics: list[Nic] = field(default_factory=list)
def json(self) -> dict[str, Any]: def json(self) -> _NetworkConfigurationSerialization:
config: dict[str, Any] = {'type': self.type.value} config: _NetworkConfigurationSerialization = {'type': self.type.value}
if self.nics: if self.nics:
config['nics'] = [n.json() for n in self.nics] config['nics'] = [n.json() for n in self.nics]
return config return config
@staticmethod @staticmethod
def parse_arg(config: dict[str, Any]) -> NetworkConfiguration | None: def parse_arg(config: _NetworkConfigurationSerialization) -> NetworkConfiguration | None:
nic_type = config.get('type', None) nic_type = config.get('type', None)
if not nic_type: if not nic_type:
return None return None
@ -126,7 +140,7 @@ class NetworkConfiguration:
def install_network_config( def install_network_config(
self, self,
installation: Any, installation: Installer,
profile_config: ProfileConfiguration | None = None profile_config: ProfileConfiguration | None = None
) -> None: ) -> None:
match self.type: match self.type:

View File

@ -1,12 +1,21 @@
from __future__ import annotations from __future__ import annotations
from dataclasses import dataclass from dataclasses import dataclass
from typing import Any from typing import TYPE_CHECKING, TypedDict
from archinstall.default_profiles.profile import GreeterType, Profile from archinstall.default_profiles.profile import GreeterType, Profile
from ..hardware import GfxDriver from ..hardware import GfxDriver
if TYPE_CHECKING:
from archinstall.lib.profile.profiles_handler import ProfileSerialization
class _ProfileConfigurationSerialization(TypedDict):
profile: ProfileSerialization
gfx_driver: str | None
greeter: str | None
@dataclass @dataclass
class ProfileConfiguration: class ProfileConfiguration:
@ -14,7 +23,7 @@ class ProfileConfiguration:
gfx_driver: GfxDriver | None = None gfx_driver: GfxDriver | None = None
greeter: GreeterType | None = None greeter: GreeterType | None = None
def json(self) -> dict[str, Any]: def json(self) -> _ProfileConfigurationSerialization:
from ..profile.profiles_handler import profile_handler from ..profile.profiles_handler import profile_handler
return { return {
'profile': profile_handler.to_json(self.profile), 'profile': profile_handler.to_json(self.profile),
@ -23,7 +32,7 @@ class ProfileConfiguration:
} }
@classmethod @classmethod
def parse_arg(cls, arg: dict[str, Any]) -> 'ProfileConfiguration': def parse_arg(cls, arg: _ProfileConfigurationSerialization) -> 'ProfileConfiguration':
from ..profile.profiles_handler import profile_handler from ..profile.profiles_handler import profile_handler
profile = profile_handler.parse_profile_config(arg['profile']) profile = profile_handler.parse_profile_config(arg['profile'])
greeter = arg.get('greeter', None) greeter = arg.get('greeter', None)

View File

@ -115,7 +115,7 @@ class User:
# if it's every going to be used # if it's every going to be used
return [] return []
def json(self) -> dict[str, Any]: def json(self) -> dict[str, str | bool]:
return { return {
'username': self.username, 'username': self.username,
'!password': self.password, '!password': self.password,

View File

@ -13,7 +13,9 @@ from .config import Config
from .repo import Repo from .repo import Repo
if TYPE_CHECKING: if TYPE_CHECKING:
_: Any from archinstall.lib.translationhandler import DeferredTranslation
_: Callable[[str], DeferredTranslation]
class Pacman: class Pacman:
@ -33,14 +35,14 @@ class Pacman:
pacman_db_lock = Path('/var/lib/pacman/db.lck') pacman_db_lock = Path('/var/lib/pacman/db.lck')
if pacman_db_lock.exists(): if pacman_db_lock.exists():
warn(_('Pacman is already running, waiting maximum 10 minutes for it to terminate.')) warn(str(_('Pacman is already running, waiting maximum 10 minutes for it to terminate.')))
started = time.time() started = time.time()
while pacman_db_lock.exists(): while pacman_db_lock.exists():
time.sleep(0.25) time.sleep(0.25)
if time.time() - started > (60 * 10): if time.time() - started > (60 * 10):
error(_('Pre-existing pacman lock never exited. Please clean up any existing pacman sessions before using archinstall.')) error(str(_('Pre-existing pacman lock never exited. Please clean up any existing pacman sessions before using archinstall.')))
exit(1) exit(1)
return SysCommand(f'{default_cmd} {args}') return SysCommand(f'{default_cmd} {args}')

View File

@ -8,7 +8,7 @@ from functools import cached_property
from pathlib import Path from pathlib import Path
from tempfile import NamedTemporaryFile from tempfile import NamedTemporaryFile
from types import ModuleType from types import ModuleType
from typing import TYPE_CHECKING, Any from typing import TYPE_CHECKING, NotRequired, TypedDict
from ...default_profiles.profile import GreeterType, Profile from ...default_profiles.profile import GreeterType, Profile
from ..hardware import GfxDriver from ..hardware import GfxDriver
@ -26,6 +26,13 @@ if TYPE_CHECKING:
_: Callable[[str], DeferredTranslation] _: Callable[[str], DeferredTranslation]
class ProfileSerialization(TypedDict):
main: NotRequired[str]
details: NotRequired[list[str]]
custom_settings: NotRequired[dict[str, dict[str, str | None]]]
path: NotRequired[str]
class ProfileHandler: class ProfileHandler:
def __init__(self) -> None: def __init__(self) -> None:
self._profiles: list[Profile] | None = None self._profiles: list[Profile] | None = None
@ -33,13 +40,13 @@ class ProfileHandler:
# special variable to keep track of a profile url configuration # special variable to keep track of a profile url configuration
# it is merely used to be able to export the path again when a user # it is merely used to be able to export the path again when a user
# wants to save the configuration # wants to save the configuration
self._url_path = None self._url_path: str | None = None
def to_json(self, profile: Profile | None) -> dict[str, Any]: def to_json(self, profile: Profile | None) -> ProfileSerialization:
""" """
Serialize the selected profile setting to JSON Serialize the selected profile setting to JSON
""" """
data: dict[str, Any] = {} data: ProfileSerialization = {}
if profile is not None: if profile is not None:
data = { data = {
@ -53,7 +60,7 @@ class ProfileHandler:
return data return data
def parse_profile_config(self, profile_config: dict[str, Any]) -> Profile | None: def parse_profile_config(self, profile_config: ProfileSerialization) -> Profile | None:
""" """
Deserialize JSON configuration for profile Deserialize JSON configuration for profile
""" """