Remove remaining Optional and Union usage from the codebase (#2868)
This commit is contained in:
correctmost
GitHub
parent
97d6d84c3c
commit
80b4dab092
|
|
@ -7,7 +7,7 @@ import time
|
|||
import uuid
|
||||
from collections.abc import Iterable
|
||||
from pathlib import Path
|
||||
from typing import Any, Optional, TYPE_CHECKING, Literal
|
||||
from typing import Any, TYPE_CHECKING, Literal
|
||||
|
||||
from parted import (
|
||||
Disk, Geometry, FileSystem,
|
||||
|
|
@ -138,8 +138,8 @@ class DeviceHandler:
|
|||
def _determine_fs_type(
|
||||
self,
|
||||
partition: Partition,
|
||||
lsblk_info: Optional[LsblkInfo] = None
|
||||
) -> Optional[FilesystemType]:
|
||||
lsblk_info: LsblkInfo | None = None
|
||||
) -> FilesystemType | None:
|
||||
try:
|
||||
if partition.fileSystem:
|
||||
return FilesystemType(partition.fileSystem.type)
|
||||
|
|
@ -151,17 +151,17 @@ class DeviceHandler:
|
|||
|
||||
return None
|
||||
|
||||
def get_device(self, path: Path) -> Optional[BDevice]:
|
||||
def get_device(self, path: Path) -> BDevice | None:
|
||||
return self._devices.get(path, None)
|
||||
|
||||
def get_device_by_partition_path(self, partition_path: Path) -> Optional[BDevice]:
|
||||
def get_device_by_partition_path(self, partition_path: Path) -> BDevice | None:
|
||||
partition = self.find_partition(partition_path)
|
||||
if partition:
|
||||
device: Device = partition.disk.device
|
||||
return self.get_device(Path(device.path))
|
||||
return None
|
||||
|
||||
def find_partition(self, path: Path) -> Optional[_PartitionInfo]:
|
||||
def find_partition(self, path: Path) -> _PartitionInfo | None:
|
||||
for device in self._devices.values():
|
||||
part = next(filter(lambda x: str(x.path) == str(path), device.partition_infos), None)
|
||||
if part is not None:
|
||||
|
|
@ -172,7 +172,7 @@ class DeviceHandler:
|
|||
lsblk = get_lsblk_info(dev_path)
|
||||
return Path(f'/dev/{lsblk.pkname}')
|
||||
|
||||
def get_unique_path_for_device(self, dev_path: Path) -> Optional[Path]:
|
||||
def get_unique_path_for_device(self, dev_path: Path) -> Path | None:
|
||||
paths = Path('/dev/disk/by-id').glob('*')
|
||||
linked_targets = {p.resolve(): p for p in paths}
|
||||
linked_wwn_targets = {
|
||||
|
|
@ -188,14 +188,14 @@ class DeviceHandler:
|
|||
|
||||
return None
|
||||
|
||||
def get_uuid_for_path(self, path: Path) -> Optional[str]:
|
||||
def get_uuid_for_path(self, path: Path) -> str | None:
|
||||
partition = self.find_partition(path)
|
||||
return partition.partuuid if partition else None
|
||||
|
||||
def get_btrfs_info(
|
||||
self,
|
||||
dev_path: Path,
|
||||
lsblk_info: Optional[LsblkInfo] = None
|
||||
lsblk_info: LsblkInfo | None = None
|
||||
) -> list[_BtrfsSubvolumeInfo]:
|
||||
if not lsblk_info:
|
||||
lsblk_info = get_lsblk_info(dev_path)
|
||||
|
|
@ -286,7 +286,7 @@ class DeviceHandler:
|
|||
def encrypt(
|
||||
self,
|
||||
dev_path: Path,
|
||||
mapper_name: Optional[str],
|
||||
mapper_name: str | None,
|
||||
enc_password: str,
|
||||
lock_after_create: bool = True
|
||||
) -> Luks2:
|
||||
|
|
@ -312,7 +312,7 @@ class DeviceHandler:
|
|||
def format_encrypted(
|
||||
self,
|
||||
dev_path: Path,
|
||||
mapper_name: Optional[str],
|
||||
mapper_name: str | None,
|
||||
fs_type: FilesystemType,
|
||||
enc_conf: DiskEncryption
|
||||
) -> None:
|
||||
|
|
@ -339,7 +339,7 @@ class DeviceHandler:
|
|||
self,
|
||||
cmd: str,
|
||||
info_type: Literal['lv', 'vg', 'pvseg']
|
||||
) -> Optional[Any]:
|
||||
) -> Any | None:
|
||||
raw_info = SysCommand(cmd).decode().split('\n')
|
||||
|
||||
# for whatever reason the output sometimes contains
|
||||
|
|
@ -377,14 +377,14 @@ class DeviceHandler:
|
|||
|
||||
return None
|
||||
|
||||
def _lvm_info_with_retry(self, cmd: str, info_type: Literal['lv', 'vg', 'pvseg']) -> Optional[Any]:
|
||||
def _lvm_info_with_retry(self, cmd: str, info_type: Literal['lv', 'vg', 'pvseg']) -> Any | None:
|
||||
while True:
|
||||
try:
|
||||
return self._lvm_info(cmd, info_type)
|
||||
except ValueError:
|
||||
time.sleep(3)
|
||||
|
||||
def lvm_vol_info(self, lv_name: str) -> Optional[LvmVolumeInfo]:
|
||||
def lvm_vol_info(self, lv_name: str) -> LvmVolumeInfo | None:
|
||||
cmd = (
|
||||
'lvs --reportformat json '
|
||||
'--unit B '
|
||||
|
|
@ -393,7 +393,7 @@ class DeviceHandler:
|
|||
|
||||
return self._lvm_info_with_retry(cmd, 'lv')
|
||||
|
||||
def lvm_group_info(self, vg_name: str) -> Optional[LvmGroupInfo]:
|
||||
def lvm_group_info(self, vg_name: str) -> LvmGroupInfo | None:
|
||||
cmd = (
|
||||
'vgs --reportformat json '
|
||||
'--unit B '
|
||||
|
|
@ -403,7 +403,7 @@ class DeviceHandler:
|
|||
|
||||
return self._lvm_info_with_retry(cmd, 'vg')
|
||||
|
||||
def lvm_pvseg_info(self, vg_name: str, lv_name: str) -> Optional[LvmPVInfo]:
|
||||
def lvm_pvseg_info(self, vg_name: str, lv_name: str) -> LvmPVInfo | None:
|
||||
cmd = (
|
||||
'pvs '
|
||||
'--segments -o+lv_name,vg_name '
|
||||
|
|
@ -457,7 +457,7 @@ class DeviceHandler:
|
|||
worker.poll()
|
||||
worker.write(b'y\n', line_ending=False)
|
||||
|
||||
def lvm_vol_create(self, vg_name: str, volume: LvmVolume, offset: Optional[Size] = None) -> None:
|
||||
def lvm_vol_create(self, vg_name: str, volume: LvmVolume, offset: Size | None = None) -> None:
|
||||
if offset is not None:
|
||||
length = volume.length - offset
|
||||
else:
|
||||
|
|
@ -593,7 +593,7 @@ class DeviceHandler:
|
|||
def create_btrfs_volumes(
|
||||
self,
|
||||
part_mod: PartitionModification,
|
||||
enc_conf: Optional['DiskEncryption'] = None
|
||||
enc_conf: 'DiskEncryption | None' = None
|
||||
) -> None:
|
||||
info(f'Creating subvolumes: {part_mod.safe_dev_path}')
|
||||
|
||||
|
|
@ -663,7 +663,7 @@ class DeviceHandler:
|
|||
def partition(
|
||||
self,
|
||||
modification: DeviceModification,
|
||||
partition_table: Optional[PartitionTable] = None
|
||||
partition_table: PartitionTable | None = None
|
||||
) -> None:
|
||||
"""
|
||||
Create a partition table on the block device and create all partitions.
|
||||
|
|
@ -701,7 +701,7 @@ class DeviceHandler:
|
|||
self,
|
||||
dev_path: Path,
|
||||
target_mountpoint: Path,
|
||||
mount_fs: Optional[str] = None,
|
||||
mount_fs: str | None = None,
|
||||
create_target_mountpoint: bool = True,
|
||||
options: list[str] = []
|
||||
) -> None:
|
||||
|
|
@ -777,7 +777,7 @@ class DeviceHandler:
|
|||
|
||||
return device_mods
|
||||
|
||||
def partprobe(self, path: Optional[Path] = None) -> None:
|
||||
def partprobe(self, path: Path | None = None) -> None:
|
||||
if path is not None:
|
||||
command = f'partprobe {path}'
|
||||
else:
|
||||
|
|
|
|||
|
|
@ -6,8 +6,7 @@ import uuid
|
|||
from dataclasses import dataclass, field
|
||||
from enum import Enum
|
||||
from pathlib import Path
|
||||
from typing import Optional, TYPE_CHECKING, Any
|
||||
from typing import Union
|
||||
from typing import TYPE_CHECKING, Any
|
||||
|
||||
import parted
|
||||
from parted import Disk, Geometry, Partition
|
||||
|
|
@ -41,10 +40,10 @@ class DiskLayoutType(Enum):
|
|||
class DiskLayoutConfiguration:
|
||||
config_type: DiskLayoutType
|
||||
device_modifications: list[DeviceModification] = field(default_factory=list)
|
||||
lvm_config: Optional[LvmConfiguration] = None
|
||||
lvm_config: LvmConfiguration | None = None
|
||||
|
||||
# used for pre-mounted config
|
||||
mountpoint: Optional[Path] = None
|
||||
mountpoint: Path | None = None
|
||||
|
||||
def json(self) -> dict[str, Any]:
|
||||
if self.config_type == DiskLayoutType.Pre_mount:
|
||||
|
|
@ -64,7 +63,7 @@ class DiskLayoutConfiguration:
|
|||
return config
|
||||
|
||||
@classmethod
|
||||
def parse_arg(cls, disk_config: dict[str, Any]) -> Optional[DiskLayoutConfiguration]:
|
||||
def parse_arg(cls, disk_config: dict[str, Any]) -> DiskLayoutConfiguration | None:
|
||||
from .device_handler import device_handler
|
||||
|
||||
device_modifications: list[DeviceModification] = []
|
||||
|
|
@ -238,7 +237,7 @@ class Size:
|
|||
def convert(
|
||||
self,
|
||||
target_unit: Unit,
|
||||
sector_size: Optional[SectorSize] = None
|
||||
sector_size: SectorSize | None = None
|
||||
) -> Size:
|
||||
if target_unit == Unit.sectors and sector_size is None:
|
||||
raise ValueError('If target has unit sector, a sector size must be provided')
|
||||
|
|
@ -266,7 +265,7 @@ class Size:
|
|||
def format_size(
|
||||
self,
|
||||
target_unit: Unit,
|
||||
sector_size: Optional[SectorSize] = None,
|
||||
sector_size: SectorSize | None = None,
|
||||
include_unit: bool = True
|
||||
) -> str:
|
||||
target_size = self.convert(target_unit, sector_size)
|
||||
|
|
@ -333,7 +332,7 @@ class BtrfsMountOption(Enum):
|
|||
@dataclass
|
||||
class _BtrfsSubvolumeInfo:
|
||||
name: Path
|
||||
mountpoint: Optional[Path]
|
||||
mountpoint: Path | None
|
||||
|
||||
|
||||
@dataclass
|
||||
|
|
@ -341,14 +340,14 @@ class _PartitionInfo:
|
|||
partition: Partition
|
||||
name: str
|
||||
type: PartitionType
|
||||
fs_type: Optional[FilesystemType]
|
||||
fs_type: FilesystemType | None
|
||||
path: Path
|
||||
start: Size
|
||||
length: Size
|
||||
flags: list[PartitionFlag]
|
||||
partn: Optional[int]
|
||||
partuuid: Optional[str]
|
||||
uuid: Optional[str]
|
||||
partn: int | None
|
||||
partuuid: str | None
|
||||
uuid: str | None
|
||||
disk: Disk
|
||||
mountpoints: list[Path]
|
||||
btrfs_subvol_infos: list[_BtrfsSubvolumeInfo] = field(default_factory=list)
|
||||
|
|
@ -381,10 +380,10 @@ class _PartitionInfo:
|
|||
def from_partition(
|
||||
cls,
|
||||
partition: Partition,
|
||||
fs_type: Optional[FilesystemType],
|
||||
partn: Optional[int],
|
||||
partuuid: Optional[str],
|
||||
uuid: Optional[str],
|
||||
fs_type: FilesystemType | None,
|
||||
partn: int | None,
|
||||
partuuid: str | None,
|
||||
uuid: str | None,
|
||||
mountpoints: list[Path],
|
||||
btrfs_subvol_infos: list[_BtrfsSubvolumeInfo] = []
|
||||
) -> _PartitionInfo:
|
||||
|
|
@ -473,7 +472,7 @@ class _DeviceInfo:
|
|||
@dataclass
|
||||
class SubvolumeModification:
|
||||
name: Path
|
||||
mountpoint: Optional[Path] = None
|
||||
mountpoint: Path | None = None
|
||||
|
||||
@classmethod
|
||||
def from_existing_subvol_info(cls, info: _BtrfsSubvolumeInfo) -> SubvolumeModification:
|
||||
|
|
@ -572,7 +571,7 @@ class PartitionType(Enum):
|
|||
debug(f'Partition code not supported: {code}')
|
||||
return PartitionType._Unknown
|
||||
|
||||
def get_partition_code(self) -> Optional[int]:
|
||||
def get_partition_code(self) -> int | None:
|
||||
if self == PartitionType.Primary:
|
||||
return parted.PARTITION_NORMAL
|
||||
elif self == PartitionType.Boot:
|
||||
|
|
@ -623,7 +622,7 @@ class FilesystemType(Enum):
|
|||
return self.value
|
||||
|
||||
@property
|
||||
def installation_pkg(self) -> Optional[str]:
|
||||
def installation_pkg(self) -> str | None:
|
||||
match self:
|
||||
case FilesystemType.Btrfs:
|
||||
return 'btrfs-progs'
|
||||
|
|
@ -635,7 +634,7 @@ class FilesystemType(Enum):
|
|||
return None
|
||||
|
||||
@property
|
||||
def installation_module(self) -> Optional[str]:
|
||||
def installation_module(self) -> str | None:
|
||||
match self:
|
||||
case FilesystemType.Btrfs:
|
||||
return 'btrfs'
|
||||
|
|
@ -643,7 +642,7 @@ class FilesystemType(Enum):
|
|||
return None
|
||||
|
||||
@property
|
||||
def installation_binary(self) -> Optional[str]:
|
||||
def installation_binary(self) -> str | None:
|
||||
match self:
|
||||
case FilesystemType.Btrfs:
|
||||
return '/usr/bin/btrfs'
|
||||
|
|
@ -651,7 +650,7 @@ class FilesystemType(Enum):
|
|||
return None
|
||||
|
||||
@property
|
||||
def installation_hooks(self) -> Optional[str]:
|
||||
def installation_hooks(self) -> str | None:
|
||||
match self:
|
||||
case FilesystemType.Btrfs:
|
||||
return 'btrfs'
|
||||
|
|
@ -672,17 +671,17 @@ class PartitionModification:
|
|||
type: PartitionType
|
||||
start: Size
|
||||
length: Size
|
||||
fs_type: Optional[FilesystemType] = None
|
||||
mountpoint: Optional[Path] = None
|
||||
fs_type: FilesystemType | None = None
|
||||
mountpoint: Path | None = None
|
||||
mount_options: list[str] = field(default_factory=list)
|
||||
flags: list[PartitionFlag] = field(default_factory=list)
|
||||
btrfs_subvols: list[SubvolumeModification] = field(default_factory=list)
|
||||
|
||||
# only set if the device was created or exists
|
||||
dev_path: Optional[Path] = None
|
||||
partn: Optional[int] = None
|
||||
partuuid: Optional[str] = None
|
||||
uuid: Optional[str] = None
|
||||
dev_path: Path | None = None
|
||||
partn: int | None = None
|
||||
partuuid: str | None = None
|
||||
uuid: str | None = None
|
||||
|
||||
_efi_indicator_flags = (PartitionFlag.Boot, PartitionFlag.ESP)
|
||||
_boot_indicator_flags = (PartitionFlag.Boot, PartitionFlag.XBOOTLDR)
|
||||
|
|
@ -798,7 +797,7 @@ class PartitionModification:
|
|||
return self.status in [ModificationStatus.Create, ModificationStatus.Modify]
|
||||
|
||||
@property
|
||||
def mapper_name(self) -> Optional[str]:
|
||||
def mapper_name(self) -> str | None:
|
||||
if self.dev_path:
|
||||
return f'{storage.get("ENC_IDENTIFIER", "ai")}{self.dev_path.name}'
|
||||
return None
|
||||
|
|
@ -913,14 +912,14 @@ class LvmVolume:
|
|||
name: str
|
||||
fs_type: FilesystemType
|
||||
length: Size
|
||||
mountpoint: Optional[Path]
|
||||
mountpoint: Path | None
|
||||
mount_options: list[str] = field(default_factory=list)
|
||||
btrfs_subvols: list[SubvolumeModification] = field(default_factory=list)
|
||||
|
||||
# volume group name
|
||||
vg_name: Optional[str] = None
|
||||
vg_name: str | None = None
|
||||
# mapper device path /dev/<vg>/<vol>
|
||||
dev_path: Optional[Path] = None
|
||||
dev_path: Path | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
# needed to use the object as a dictionary key due to hash func
|
||||
|
|
@ -937,7 +936,7 @@ class LvmVolume:
|
|||
return ''
|
||||
|
||||
@property
|
||||
def mapper_name(self) -> Optional[str]:
|
||||
def mapper_name(self) -> str | None:
|
||||
if self.dev_path:
|
||||
return f'{storage.get("ENC_IDENTIFIER", "ai")}{self.safe_dev_path.name}'
|
||||
return None
|
||||
|
|
@ -1100,7 +1099,7 @@ class LvmConfiguration:
|
|||
|
||||
return volumes
|
||||
|
||||
def get_root_volume(self) -> Optional[LvmVolume]:
|
||||
def get_root_volume(self) -> LvmVolume | None:
|
||||
for vg in self.vol_groups:
|
||||
filtered = next(filter(lambda x: x.is_root(), vg.volumes), None)
|
||||
if filtered:
|
||||
|
|
@ -1131,14 +1130,14 @@ class DeviceModification:
|
|||
def add_partition(self, partition: PartitionModification) -> None:
|
||||
self.partitions.append(partition)
|
||||
|
||||
def get_efi_partition(self) -> Optional[PartitionModification]:
|
||||
def get_efi_partition(self) -> PartitionModification | None:
|
||||
"""
|
||||
Similar to get_boot_partition() but excludes XBOOTLDR partitions from it's candidates.
|
||||
"""
|
||||
filtered = filter(lambda x: x.is_efi() and x.mountpoint, self.partitions)
|
||||
return next(filtered, None)
|
||||
|
||||
def get_boot_partition(self) -> Optional[PartitionModification]:
|
||||
def get_boot_partition(self) -> PartitionModification | None:
|
||||
"""
|
||||
Returns the first partition marked as XBOOTLDR (PARTTYPE id of bc13c2ff-...) or Boot and has a mountpoint.
|
||||
Only returns XBOOTLDR if separate EFI is detected using self.get_efi_partition()
|
||||
|
|
@ -1153,7 +1152,7 @@ class DeviceModification:
|
|||
filtered = filter(lambda x: x.is_boot() and x.mountpoint, self.partitions)
|
||||
return next(filtered, None)
|
||||
|
||||
def get_root_partition(self) -> Optional[PartitionModification]:
|
||||
def get_root_partition(self) -> PartitionModification | None:
|
||||
filtered = filter(lambda x: x.is_root(), self.partitions)
|
||||
return next(filtered, None)
|
||||
|
||||
|
|
@ -1201,7 +1200,7 @@ class DiskEncryption:
|
|||
encryption_password: str = ''
|
||||
partitions: list[PartitionModification] = field(default_factory=list)
|
||||
lvm_volumes: list[LvmVolume] = field(default_factory=list)
|
||||
hsm_device: Optional[Fido2Device] = None
|
||||
hsm_device: Fido2Device | None = None
|
||||
|
||||
def __post_init__(self) -> None:
|
||||
if self.encryption_type in [EncryptionType.Luks, EncryptionType.LvmOnLuks] and not self.partitions:
|
||||
|
|
@ -1249,7 +1248,7 @@ class DiskEncryption:
|
|||
disk_config: DiskLayoutConfiguration,
|
||||
disk_encryption: dict[str, Any],
|
||||
password: str = ''
|
||||
) -> Optional['DiskEncryption']:
|
||||
) -> 'DiskEncryption | None':
|
||||
if not cls.validate_enc(disk_config):
|
||||
return None
|
||||
|
||||
|
|
@ -1363,7 +1362,7 @@ class LsblkOutput(BaseModel):
|
|||
|
||||
|
||||
def _fetch_lsblk_info(
|
||||
dev_path: Optional[Union[Path, str]] = None,
|
||||
dev_path: Path | str | None = None,
|
||||
reverse: bool = False,
|
||||
full_dev_path: bool = False
|
||||
) -> list[LsblkInfo]:
|
||||
|
|
@ -1401,7 +1400,7 @@ def _fetch_lsblk_info(
|
|||
|
||||
|
||||
def get_lsblk_info(
|
||||
dev_path: Union[Path, str],
|
||||
dev_path: Path | str,
|
||||
reverse: bool = False,
|
||||
full_dev_path: bool = False
|
||||
) -> LsblkInfo:
|
||||
|
|
@ -1416,9 +1415,9 @@ def get_all_lsblk_info() -> list[LsblkInfo]:
|
|||
|
||||
|
||||
def find_lsblk_info(
|
||||
dev_path: Union[Path, str],
|
||||
dev_path: Path | str,
|
||||
info: list[LsblkInfo]
|
||||
) -> Optional[LsblkInfo]:
|
||||
) -> LsblkInfo | None:
|
||||
if isinstance(dev_path, str):
|
||||
dev_path = Path(dev_path)
|
||||
|
||||
|
|
|
|||
|
|
@ -194,7 +194,6 @@ select = [
|
|||
ignore = [
|
||||
"E722", # bare-except
|
||||
"PLW2901", # redefined-loop-name
|
||||
"UP007", # non-pep604-annotation
|
||||
"UP027", # unpacked-list-comprehension
|
||||
"UP028", # yield-in-for-loop
|
||||
"UP031", # printf-string-formatting
|
||||
|
|
|
|||
Loading…
Reference in New Issue