Skip to content

Commit

Permalink
Merge pull request #2334 from OSInside/more-type-hints
Browse files Browse the repository at this point in the history
Add more type hints
  • Loading branch information
Conan-Kudo authored Aug 1, 2023
2 parents bced159 + 5ad1145 commit d5dae47
Show file tree
Hide file tree
Showing 16 changed files with 122 additions and 57 deletions.
2 changes: 2 additions & 0 deletions .virtualenv.requirements.txt
Original file line number Diff line number Diff line change
Expand Up @@ -23,3 +23,5 @@ xmltodict

# json support that also knows about tuples
simplejson

typing_extensions; python_version < '3.8'
3 changes: 1 addition & 2 deletions kiwi/boot/image/builtin_kiwi.py
Original file line number Diff line number Diff line change
Expand Up @@ -199,10 +199,9 @@ def create_initrd(
compress = Compress(
os.sep.join([self.target_dir, kiwi_initrd_basename])
)
compress.xz(
self.initrd_filename = compress.xz(
['--check=crc32', '--lzma2=dict=1MiB', '--threads=0']
)
self.initrd_filename = compress.compressed_filename

def cleanup(self) -> None:
for directory in self.temp_directories:
Expand Down
2 changes: 1 addition & 1 deletion kiwi/builder/container.py
Original file line number Diff line number Diff line change
Expand Up @@ -145,7 +145,7 @@ def create(self) -> Result:
self.requested_container_type, self.root_dir, self.container_config
)
self.filename = container_image.create(
self.filename, self.base_image, self.ensure_empty_tmpdirs,
self.filename, self.base_image or '', self.ensure_empty_tmpdirs,
self.runtime_config.get_container_compression()
# appx containers already contains a compressed root
if self.requested_container_type != 'appx' else False
Expand Down
3 changes: 1 addition & 2 deletions kiwi/builder/kis.py
Original file line number Diff line number Diff line change
Expand Up @@ -124,8 +124,7 @@ def create(self) -> Result:
if self.compressed:
log.info('xz compressing root filesystem image')
compress = Compress(self.image)
compress.xz(self.xz_options)
self.image = compress.compressed_filename
self.image = compress.xz(self.xz_options)

log.info('Creating root filesystem MD5 checksum')
checksum = Checksum(self.image)
Expand Down
3 changes: 2 additions & 1 deletion kiwi/container/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -22,6 +22,7 @@
)
from typing import Dict

from kiwi.container.base import ContainerImageBase
from kiwi.exceptions import (
KiwiContainerImageSetupError
)
Expand All @@ -40,7 +41,7 @@ def __init__(self) -> None:
return None # pragma: no cover

@staticmethod
def new(name: str, root_dir: str, custom_args: Dict=None): # noqa: E252
def new(name: str, root_dir: str, custom_args: Dict=None) -> ContainerImageBase: # noqa: E252
name_map = {
'docker': 'OCI',
'oci': 'OCI',
Expand Down
11 changes: 5 additions & 6 deletions kiwi/container/appx.py
Original file line number Diff line number Diff line change
Expand Up @@ -26,13 +26,13 @@
from kiwi.utils.compress import Compress
from kiwi.runtime_config import RuntimeConfig
from kiwi.command import Command

from kiwi.container.base import ContainerImageBase
from kiwi.exceptions import KiwiContainerSetupError

log = logging.getLogger('kiwi')


class ContainerImageAppx:
class ContainerImageAppx(ContainerImageBase):
"""
Create Appx container from a root directory for
WSL(Windows Subsystem Linux)
Expand All @@ -50,7 +50,7 @@ class ContainerImageAppx:
'metadata_path': 'directory'
}
"""
def __init__(self, root_dir: str, custom_args: Dict = {}):
def __init__(self, root_dir: str, custom_args: Dict[str, str] = {}):
self.root_dir = root_dir
self.wsl_config = custom_args or {}
self.runtime_config = RuntimeConfig()
Expand All @@ -72,7 +72,7 @@ def __init__(self, root_dir: str, custom_args: Dict = {}):
def create(
self, filename: str, base_image: str = '',
ensure_empty_tmpdirs: bool = False, compress_archive: bool = False
):
) -> str:
"""
Create WSL/Appx archive
Expand Down Expand Up @@ -126,7 +126,6 @@ def create(
)
if compress_archive:
compress = Compress(filename)
compress.xz(self.runtime_config.get_xz_options())
filename = compress.compressed_filename
filename = compress.xz(self.runtime_config.get_xz_options())

return filename
27 changes: 27 additions & 0 deletions kiwi/container/base.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,27 @@
# Copyright (c) 2023 SUSE LLC
#
# This file is part of kiwi.
#
# kiwi is free software: you can redistribute it and/or modify
# it under the terms of the GNU General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# kiwi is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU General Public License for more details.
#
# You should have received a copy of the GNU General Public License
# along with kiwi. If not, see <http://www.gnu.org/licenses/>
#
from abc import ABC, abstractmethod


class ContainerImageBase(ABC):
@abstractmethod
def create(
self, filename: str, base_image: str,
ensure_empty_tmpdirs: bool, compress_archive: bool = False
) -> str:
pass # pragma: no cover
41 changes: 29 additions & 12 deletions kiwi/container/oci.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,20 +17,40 @@
#
import os
import logging
from typing import (
Dict, List
)
import sys
from typing import Dict, List, Optional
if sys.version_info >= (3, 8):
from typing import TypedDict # pragma: no cover
else: # pragma: no cover
from typing_extensions import TypedDict # pragma: no cover

# project
from kiwi.utils.compress import Compress
from kiwi.runtime_config import RuntimeConfig
from kiwi.defaults import Defaults
from kiwi.oci_tools import OCI
from kiwi.container.base import ContainerImageBase

log = logging.getLogger('kiwi')


class ContainerImageOCI:
class OciConfig(TypedDict, total=False):
container_name: str
container_tag: str
additional_names: List[str]
entry_command: List[str]
entry_subcommand: List[str]
maintainer: str
user: str
workingdir: str
expose_ports: List[str]
volumes: List[str]
environment: Dict[str, str]
labels: Dict[str, str]
history: Dict[str, str]


class ContainerImageOCI(ContainerImageBase):
"""
Create oci container from a root directory
Expand Down Expand Up @@ -63,13 +83,10 @@ class ContainerImageOCI:
}
}
"""
def __init__(self, root_dir: str, transport: str, custom_args: Dict = {}):
def __init__(self, root_dir: str, transport: str, custom_args: Optional[OciConfig] = None) -> None:
self.root_dir = root_dir
self.archive_transport = transport
if custom_args:
self.oci_config = custom_args
else:
self.oci_config = {}
self.oci_config = custom_args or {}

# for builds inside the buildservice we include a reference to the
# specific build. Thus disturl label only exists inside the
Expand Down Expand Up @@ -102,10 +119,11 @@ def __init__(self, root_dir: str, transport: str, custom_args: Dict = {}):
self.oci_config['history']['created_by'] = \
Defaults.get_default_container_created_by()


def create(
self, filename: str, base_image: str,
ensure_empty_tmpdirs: bool = True, compress_archive: bool = False
):
) -> str:
"""
Create compressed oci system container tar archive
Expand Down Expand Up @@ -174,8 +192,7 @@ def create(
)
if compress_archive:
compress = Compress(filename)
compress.xz(RuntimeConfig().get_xz_options())
filename = compress.compressed_filename
filename = compress.xz(RuntimeConfig().get_xz_options())

return filename

Expand Down
11 changes: 7 additions & 4 deletions kiwi/iso_tools/xorriso.py
Original file line number Diff line number Diff line change
Expand Up @@ -17,14 +17,14 @@
#
import os
from typing import (
Dict, List
Dict, List, Optional
)

# project
from kiwi.iso_tools.base import IsoToolsBase
from kiwi.path import Path
from kiwi.command import Command
from kiwi.exceptions import KiwiIsoToolError
from kiwi.exceptions import KiwiFileNotFound, KiwiIsoToolError
from kiwi.defaults import Defaults


Expand Down Expand Up @@ -63,7 +63,7 @@ def get_tool_name(self) -> str:
raise KiwiIsoToolError('xorriso tool not found')

def init_iso_creation_parameters(
self, custom_args: Dict[str, str] = None
self, custom_args: Optional[Dict[str, str]] = None
) -> None:
"""
Create a set of standard parameters
Expand Down Expand Up @@ -113,9 +113,12 @@ def init_iso_creation_parameters(
]
else:
loader_file = self.boot_path + '/loader/isolinux.bin'
mbr_file = Path.which(
mbr_file_c = Path.which(
'isohdpfx.bin', Defaults.get_syslinux_search_paths()
)
if not mbr_file_c:
raise KiwiFileNotFound("isohdpfx.bin not found in the systlinux search paths")
mbr_file = mbr_file_c
self.iso_loaders += [
'-boot_image', 'isolinux', 'bin_path={0}'.format(
loader_file
Expand Down
8 changes: 3 additions & 5 deletions kiwi/package_manager/pacman.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,7 +18,7 @@
import re
import logging
from typing import (
List, Dict
List, Dict, Optional
)

# project
Expand All @@ -42,17 +42,15 @@ class PackageManagerPacman(PackageManagerBase):
pacman command environment from repository
runtime configuration
"""
def post_init(self, custom_args: List = None) -> None:
def post_init(self, custom_args: Optional[List[str]] = None) -> None:
"""
Post initialization method
Store custom pacman arguments
:param list custom_args: custom pacman arguments
"""
self.custom_args = custom_args
if not custom_args:
self.custom_args = []
self.custom_args: List[str] = custom_args or []

runtime_config = self.repository.runtime_config()
self.pacman_args = runtime_config['pacman_args']
Expand Down
29 changes: 16 additions & 13 deletions kiwi/path.py
Original file line number Diff line number Diff line change
Expand Up @@ -18,6 +18,7 @@
import os
import logging
import collections
from typing import Dict, List, Optional

# project
from kiwi.command import Command
Expand All @@ -31,7 +32,7 @@ class Path:
**Directory path helpers**
"""
@staticmethod
def sort_by_hierarchy(path_list):
def sort_by_hierarchy(path_list: List[str]) -> List[str]:
"""
Sort given list of path names by their hierachy in the tree
Expand All @@ -47,7 +48,7 @@ def sort_by_hierarchy(path_list):
:rtype: list
"""
paths_at_depth = {}
paths_at_depth: Dict[int, List[str]] = {}
for path in path_list:
path_elements = path.split('/')
path_depth = len(path_elements)
Expand All @@ -64,7 +65,7 @@ def sort_by_hierarchy(path_list):
return ordered_paths

@staticmethod
def access(path, mode, **kwargs):
def access(path: str, mode: int, **kwargs) -> bool:
"""
Check whether path can be accessed with the given mode.
Expand Down Expand Up @@ -98,7 +99,7 @@ def access(path, mode, **kwargs):
return os.access(path, mode, **kwargs)

@staticmethod
def create(path):
def create(path: str) -> None:
"""
Create path and all sub directories to target
Expand All @@ -110,7 +111,7 @@ def create(path):
)

@staticmethod
def wipe(path):
def wipe(path: str) -> None:
"""
Delete path and all contents
Expand All @@ -122,7 +123,7 @@ def wipe(path):
)

@staticmethod
def remove(path):
def remove(path: str) -> None:
"""
Delete empty path, causes an error if target is not empty
Expand All @@ -133,7 +134,7 @@ def remove(path):
)

@staticmethod
def rename(cur, new):
def rename(cur: str, new: str) -> None:
"""
Move path from cur name to new name
Expand All @@ -145,7 +146,7 @@ def rename(cur, new):
)

@staticmethod
def remove_hierarchy(root, path):
def remove_hierarchy(root: str, path: str) -> None:
"""
Recursively remove an empty path and its sub directories
starting at a given root directory. Ignore non empty or
Expand Down Expand Up @@ -182,7 +183,7 @@ def remove_hierarchy(root, path):
)

@staticmethod
def move_to_root(root, elements):
def move_to_root(root: str, elements: List[str]) -> List[str]:
"""
Change the given path elements to a new root directory
Expand All @@ -204,7 +205,7 @@ def move_to_root(root, elements):
return result

@staticmethod
def rebase_to_root(root, elements):
def rebase_to_root(root: str, elements: List[str]) -> List[str]:
"""
Include the root prefix for the given paths elements
Expand All @@ -222,9 +223,10 @@ def rebase_to_root(root, elements):

@staticmethod
def which(
filename, alternative_lookup_paths=None,
custom_env=None, access_mode=None, root_dir=None
):
filename: str, alternative_lookup_paths: Optional[List[str]]=None,
custom_env: Optional[Dict[str, str]]=None, access_mode: Optional[int]=None,
root_dir: Optional[str]=None
) -> Optional[str]:
"""
Lookup file name in PATH
Expand Down Expand Up @@ -267,3 +269,4 @@ def which(
return location

log.debug(' '.join(multipart_message))
return None
Loading

0 comments on commit d5dae47

Please sign in to comment.