Fix some mypy warnings in archinstall/lib/ (#3103)

This commit is contained in:
correctmost 2025-01-11 04:44:43 +00:00 committed by GitHub
parent 22b410d082
commit e7f2a8c203
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
12 changed files with 37 additions and 21 deletions

View File

@ -326,7 +326,7 @@ class ArchConfigHandler:
return path.read_text() return path.read_text()
def _cleanup_config(self, config: Namespace | dict) -> dict[str, Any]: def _cleanup_config(self, config: Namespace | dict[str, Any]) -> dict[str, Any]:
clean_args = {} clean_args = {}
for key, val in config.items(): for key, val in config.items():
if isinstance(val, dict): if isinstance(val, dict):

View File

@ -20,7 +20,7 @@ if TYPE_CHECKING:
class ConfigurationOutput: class ConfigurationOutput:
def __init__(self, config: dict): def __init__(self, config: dict[str, Any]):
""" """
Configuration output handler to parse the existing configuration data structure and prepare for output on the Configuration output handler to parse the existing configuration data structure and prepare for output on the
console and for saving it to configuration files console and for saving it to configuration files

View File

@ -439,11 +439,17 @@ class Size:
return self._normalize() <= other._normalize() return self._normalize() <= other._normalize()
@override @override
def __eq__(self, other) -> bool: def __eq__(self, other: object) -> bool:
if not isinstance(other, Size):
return NotImplemented
return self._normalize() == other._normalize() return self._normalize() == other._normalize()
@override @override
def __ne__(self, other) -> bool: def __ne__(self, other: object) -> bool:
if not isinstance(other, Size):
return NotImplemented
return self._normalize() != other._normalize() return self._normalize() != other._normalize()
def __gt__(self, other: Size) -> bool: def __gt__(self, other: Size) -> bool:

View File

@ -477,7 +477,7 @@ def run_custom_user_commands(commands: list[str], installation: Installer) -> No
os.unlink(chroot_path) os.unlink(chroot_path)
def json_stream_to_structure(configuration_identifier: str, stream: str, target: dict) -> bool: def json_stream_to_structure(configuration_identifier: str, stream: str, target: dict[str, Any]) -> bool:
""" """
Load a JSON encoded dictionary from a stream and merge it into an existing dictionary. Load a JSON encoded dictionary from a stream and merge it into an existing dictionary.
A stream can be a filepath, a URL or a raw JSON string. A stream can be a filepath, a URL or a raw JSON string.

View File

@ -72,7 +72,7 @@ class Installer:
if accessibility_tools_in_use(): if accessibility_tools_in_use():
self._base_packages.extend(__accessibility_packages__) self._base_packages.extend(__accessibility_packages__)
self.post_base_install: list[Callable] = [] self.post_base_install: list[Callable] = [] # type: ignore[type-arg]
# TODO: Figure out which one of these two we'll use.. But currently we're mixing them.. # TODO: Figure out which one of these two we'll use.. But currently we're mixing them..
storage['session'] = self storage['session'] = self

View File

@ -40,7 +40,10 @@ class PackageSearchResult:
return self.pkgver return self.pkgver
@override @override
def __eq__(self, other) -> bool: def __eq__(self, other: object) -> bool:
if not isinstance(other, PackageSearchResult):
return NotImplemented
return self.pkg_version == other.pkg_version return self.pkg_version == other.pkg_version
def __lt__(self, other: 'PackageSearchResult') -> bool: def __lt__(self, other: 'PackageSearchResult') -> bool:
@ -99,7 +102,10 @@ class LocalPackage:
return self.version return self.version
@override @override
def __eq__(self, other) -> bool: def __eq__(self, other: object) -> bool:
if not isinstance(other, LocalPackage):
return NotImplemented
return self.pkg_version == other.pkg_version return self.pkg_version == other.pkg_version
def __lt__(self, other: 'LocalPackage') -> bool: def __lt__(self, other: 'LocalPackage') -> bool:

View File

@ -57,6 +57,7 @@ class MirrorStatusEntryV3(BaseModel):
with urllib.request.urlopen(req, None, 5) as handle, DownloadTimer(timeout=5) as timer: with urllib.request.urlopen(req, None, 5) as handle, DownloadTimer(timeout=5) as timer:
size = len(handle.read()) size = len(handle.read())
assert timer.time is not None
self._speed = size / timer.time self._speed = size / timer.time
debug(f" speed: {self._speed} ({int(self._speed / 1024 / 1024 * 100) / 100}MiB/s)") debug(f" speed: {self._speed} ({int(self._speed / 1024 / 1024 * 100) / 100}MiB/s)")
# Do not retry error # Do not retry error

View File

@ -140,7 +140,7 @@ class User:
return users return users
@classmethod @classmethod
def _parse_backwards_compatible(cls, config_users: dict, sudo: bool) -> list['User']: def _parse_backwards_compatible(cls, config_users: dict[str, dict[str, str]], sudo: bool) -> list['User']:
if len(config_users.keys()) > 0: if len(config_users.keys()) > 0:
username = list(config_users.keys())[0] username = list(config_users.keys())[0]
password = config_users[username]['!password'] password = config_users[username]['!password']
@ -153,8 +153,8 @@ class User:
@classmethod @classmethod
def parse_arguments( def parse_arguments(
cls, cls,
config_users: list[dict[str, str]] | dict[str, str], config_users: list[dict[str, str]] | dict[str, dict[str, str]],
config_superusers: list[dict[str, str]] | dict[str, str] config_superusers: list[dict[str, str]] | dict[str, dict[str, str]]
) -> list['User']: ) -> list['User']:
users = [] users = []

View File

@ -6,8 +6,8 @@ import socket
import ssl import ssl
import struct import struct
import time import time
from types import FrameType from types import FrameType, TracebackType
from typing import Any from typing import Any, Self
from urllib.error import URLError from urllib.error import URLError
from urllib.parse import urlencode from urllib.parse import urlencode
from urllib.request import urlopen from urllib.request import urlopen
@ -40,7 +40,7 @@ class DownloadTimer:
''' '''
raise DownloadTimeout(f'Download timed out after {self.timeout} second(s).') raise DownloadTimeout(f'Download timed out after {self.timeout} second(s).')
def __enter__(self): def __enter__(self) -> Self:
if self.timeout > 0: if self.timeout > 0:
self.previous_handler = signal.signal(signal.SIGALRM, self.raise_timeout) # type: ignore[assignment] self.previous_handler = signal.signal(signal.SIGALRM, self.raise_timeout) # type: ignore[assignment]
self.previous_timer = signal.alarm(self.timeout) self.previous_timer = signal.alarm(self.timeout)
@ -48,7 +48,7 @@ class DownloadTimer:
self.start_time = time.time() self.start_time = time.time()
return self return self
def __exit__(self, typ, value, traceback) -> None: def __exit__(self, typ: type[BaseException] | None, value: BaseException | None, traceback: TracebackType | None) -> None:
if self.start_time: if self.start_time:
time_delta = time.time() - self.start_time time_delta = time.time() - self.start_time
signal.alarm(0) signal.alarm(0)
@ -164,7 +164,7 @@ def build_icmp(payload: bytes) -> bytes:
return struct.pack('!BBHHH', 8, 0, checksum, 0, 1) + payload return struct.pack('!BBHHH', 8, 0, checksum, 0, 1) + payload
def ping(hostname, timeout=5) -> int: def ping(hostname, timeout: int = 5) -> int:
watchdog = select.epoll() watchdog = select.epoll()
started = time.time() started = time.time()
random_identifier = f'archinstall-{random.randint(1000, 9999)}'.encode() random_identifier = f'archinstall-{random.randint(1000, 9999)}'.encode()

View File

@ -20,7 +20,7 @@ class FormattedOutput:
def _get_values( def _get_values(
cls, cls,
o: 'DataclassInstance', o: 'DataclassInstance',
class_formatter: str | Callable | None = None, class_formatter: str | Callable | None = None, # type: ignore[type-arg]
filter_list: list[str] = [] filter_list: list[str] = []
) -> dict[str, Any]: ) -> dict[str, Any]:
""" """
@ -52,7 +52,7 @@ class FormattedOutput:
def as_table( def as_table(
cls, cls,
obj: list[Any], obj: list[Any],
class_formatter: str | Callable | None = None, class_formatter: str | Callable | None = None, # type: ignore[type-arg]
filter_list: list[str] = [], filter_list: list[str] = [],
capitalize: bool = False capitalize: bool = False
) -> str: ) -> str:

View File

@ -45,7 +45,7 @@ class Pacman:
return SysCommand(f'{default_cmd} {args}') return SysCommand(f'{default_cmd} {args}')
def ask(self, error_message: str, bail_message: str, func: Callable, *args, **kwargs) -> None: def ask(self, error_message: str, bail_message: str, func: Callable, *args, **kwargs) -> None: # type: ignore[type-arg]
while True: while True:
try: try:
func(*args, **kwargs) func(*args, **kwargs)

View File

@ -96,13 +96,16 @@ disallow_any_explicit = true
[[tool.mypy.overrides]] [[tool.mypy.overrides]]
module = "archinstall.lib.*" module = "archinstall.lib.*"
disallow_any_generics = false
disallow_any_unimported = false
disallow_incomplete_defs = false disallow_incomplete_defs = false
disallow_untyped_defs = false disallow_untyped_defs = false
warn_return_any = false warn_return_any = false
warn_unreachable = false warn_unreachable = false
[[tool.mypy.overrides]]
module = "archinstall.lib.disk.*"
# 'Any' imports are allowed because pyparted doesn't have type hints
disallow_any_unimported = false
[[tool.mypy.overrides]] [[tool.mypy.overrides]]
module = "archinstall.lib.packages" module = "archinstall.lib.packages"
disallow_any_explicit = true disallow_any_explicit = true