Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Typing and fstrings #26

Draft
wants to merge 2 commits into
base: master
Choose a base branch
from
Draft

Typing and fstrings #26

Changes from all commits
Commits
Show all changes
2 commits
Select commit Hold shift + click to select a range
f3b036c
Update typing in fablib/installer.py.
JedMeister May 9, 2022
d46558b
Use fstrings everywhere.
JedMeister May 9, 2022
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
57 changes: 27 additions & 30 deletions fablib/installer.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -11,18 +11,17 @@
import os
from os.path import join, exists, basename
import shutil
from typing import (
Iterable, Optional, Dict, Tuple, List, TextIO, IO, AnyStr, cast
)
from typing import Iterable, Optional, TextIO, IO, AnyStr, cast
import logging
logger = logging.getLogger('fab.installer')

import hashlib
import debian
from debian import debfile

from chroot import Chroot
from fablib import common

logger = logging.getLogger('fab.installer')


class Error(Exception):
pass
Expand Down Expand Up @@ -86,7 +85,7 @@ def __init__(self, path: str, lines: Iterable[str]):
class Installer:
def __init__(
self, chroot_path: str,
environ: Optional[Dict[str, str]]=None
environ: dict[str, str] = None
):
if environ is None:
environ = {}
Expand All @@ -96,7 +95,7 @@ def __init__(
self.chroot = Chroot(chroot_path, environ=env)

@staticmethod
def _get_packages_priority(packages: List[str]) -> Tuple[List[str], List[str]]:
def _get_packages_priority(packages: list[str]) -> tuple[list[str], list[str]]:
"""high priority packages must be installed before regular packages
APT should handle this, but in some circumstances it chokes...
"""
Expand All @@ -114,9 +113,9 @@ def _get_packages_priority(packages: List[str]) -> Tuple[List[str], List[str]]:
return high, regular

def _install(
self, packages: List[str],
ignore_errors: Optional[List[str]]=None,
extra_apt_args: Optional[List[str]]=None) -> None:
self, packages: list[str],
ignore_errors: list[str] = None,
extra_apt_args: list[str] = None) -> None:

if ignore_errors is None:
ignore_errors = []
Expand All @@ -139,7 +138,7 @@ def _install(
"#!/bin/sh",
"echo",
'echo "Warning: Deferring update-initramfs $@"',
'echo "update-initramfs $@" >> /%s' % defer_log,
f'echo "update-initramfs $@" >> /{defer_log}'
]

for packages in (high, regular):
Expand All @@ -150,7 +149,7 @@ def _install(
f"apt-get {' '.join((args + packages))}")
if apt_return_code != 0:

def get_last_log(path: str) -> List[str]:
def get_last_log(path: str) -> list[str]:
log = []
with open(path) as fob:
for line in fob:
Expand All @@ -163,7 +162,7 @@ def get_last_log(path: str) -> List[str]:
log.reverse()
return log

def get_errors(log: List[str], error_str: str) -> List[str]:
def get_errors(log: list[str], error_str: str) -> list[str]:
errors = []
for line in reversed(log):
if line == error_str:
Expand Down Expand Up @@ -195,13 +194,11 @@ def get_errors(log: List[str], error_str: str) -> List[str]:
errors = set(errors) - set(ignore_errors)

if ignored_errors:
print(
"Warning: ignoring package installation errors (%s)"
% " ".join(ignored_errors)
)
print(f"Warning: ignoring package installation errors"
f" ({' '.join(ignored_errors)})")

if errors:
for error in error:
for error in errors:
common.error(error)
raise Error('package installation errors')

Expand All @@ -225,15 +222,15 @@ def get_errors(log: List[str], error_str: str) -> List[str]:
os.remove(defer_log)

def install(
self, packages: List[str],
ignore_errors: Optional[List[str]]=None) -> None:
self, packages: list[str],
ignore_errors: list[str] = None) -> None:
raise NotImplementedError()


class PoolInstaller(Installer):
def __init__(
self, chroot_path: str, pool_path: str,
arch: str, environ: Optional[Dict[str, str]]=None):
arch: str, environ: dict[str, str] = None):
super(PoolInstaller, self).__init__(chroot_path, environ)

from pool_lib import Pool
Expand All @@ -244,7 +241,7 @@ def __init__(
self.arch = arch

@staticmethod
def _get_package_index(packagedir: str) -> List[str]:
def _get_package_index(packagedir: str) -> list[str]:
def filesize(path: str) -> str:
return str(os.stat(path).st_size)

Expand All @@ -262,7 +259,7 @@ def sha256sum(path: str) -> str:
# dl_path would best be calculated; but we don't have access to chroot_path here...
dl_path = os.path.join("var/cache/apt/archives", package)
if path.endswith(".deb"):
control = debian.debfile.DebFile(path).debcontrol()
control = debfile.DebFile(path).debcontrol()
for field in list(control.keys()):
index.append(field + ": " + control[field])

Expand All @@ -275,8 +272,8 @@ def sha256sum(path: str) -> str:
return index

def install(
self, packages: List[str],
ignore_errors: Optional[List[str]]=None
self, packages: list[str],
ignore_errors: list[str] = None
) -> None:
"""install packages into chroot via pool"""

Expand All @@ -297,7 +294,7 @@ def install(
print("deb file:/// local debs", file=cast(TextIO, sources_list))
sources_list.close()

index_file = "_dists_local_debs_binary-%s_Packages" % self.arch
index_file = f"_dists_local_debs_binary-{self.arch}_Packages"
index_path = join(self.chroot.path, "var/lib/apt/lists", index_file)
index = self._get_package_index(packagedir)
with open(index_path, "w") as fob:
Expand All @@ -311,15 +308,15 @@ def install(
class LiveInstaller(Installer):
def __init__(
self, chroot_path: str,
apt_proxy: Optional[str]=None,
environ: Optional[Dict[str, str]]=None):
apt_proxy: str = None,
environ: dict[str, str] = None):
super(LiveInstaller, self).__init__(chroot_path, environ)

self.apt_proxy = apt_proxy

def install(
self, packages: List[str],
ignore_errors: Optional[List[str]]=None) -> None:
self, packages: list[str],
ignore_errors: list[str] = None) -> None:
"""install packages into chroot via live apt"""
if ignore_errors is None:
ignore_errors = []
Expand Down