mirror of
ssh://git.janware.com/janware/proj/jw-pkg
synced 2026-04-28 17:15:23 +02:00
Jan Lindemann
67a2931f5e
Add the --verbose global option, which is made available as the App.verbose property. Some functions still take a verbose parameter, but the type of these parameters is converted from bool to bool|None. The idea is that, if they are None, their verbosity falls back to the global default. Signed-off-by: Jan Lindemann <jan@janware.com>
124 lines
3.2 KiB
Python
124 lines
3.2 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from __future__ import annotations
|
|
|
|
from typing import TYPE_CHECKING
|
|
|
|
if TYPE_CHECKING:
|
|
import Iterable
|
|
|
|
import abc, importlib
|
|
|
|
from .ExecContext import ExecContext, Result
|
|
from .Package import Package
|
|
from .log import *
|
|
|
|
class Distro(abc.ABC):
|
|
|
|
def __init__(self, ec: ExecContext):
|
|
assert ec is not None
|
|
self.__exec_context = ec
|
|
|
|
# == Load
|
|
|
|
@classmethod
|
|
def instantiate(cls, distro_id: str, *args, **kwargs):
|
|
backend_id = distro_id.lower().replace('-', '_')
|
|
match backend_id:
|
|
case 'ubuntu' | 'raspbian' | 'kali':
|
|
backend_id = 'debian'
|
|
case 'centos':
|
|
backend_id = 'redhat'
|
|
case 'opensuse' | 'suse':
|
|
backend_id = 'suse'
|
|
module_path = 'jw.pkg.lib.distros.' + backend_id + '.Distro'
|
|
try:
|
|
module = importlib.import_module(module_path)
|
|
except Exception as e:
|
|
log(ERR, f'Failed to import Distro module {module_path} ({str(e)})')
|
|
raise
|
|
cls = getattr(module, 'Distro')
|
|
return cls(*args, **kwargs)
|
|
|
|
# == Convenience methods
|
|
|
|
@property
|
|
def ctx(self) -> ExecContext:
|
|
return self.__exec_context
|
|
|
|
def run(self, *args, **kwargs) -> Result:
|
|
return self.__exec_context.run(*args, **kwargs)
|
|
|
|
def sudo(self, *args, **kwargs) -> Result:
|
|
return self.__exec_context.sudo(*args, **kwargs)
|
|
|
|
@property
|
|
def interactive(self) -> bool:
|
|
return self.__exec_context.interactive
|
|
|
|
# == Distribution specific methods
|
|
|
|
# -- ref
|
|
|
|
@abc.abstractmethod
|
|
async def _ref(self) -> None:
|
|
pass
|
|
|
|
async def ref(self) -> None:
|
|
return await self._ref()
|
|
|
|
# -- dup
|
|
|
|
@abc.abstractmethod
|
|
async def _dup(self, download_only: bool) -> None:
|
|
pass
|
|
|
|
async def dup(self, download_only: bool=False) -> None:
|
|
return await self._dup(download_only=download_only)
|
|
|
|
# -- reboot_required
|
|
|
|
@abc.abstractmethod
|
|
async def _reboot_required(self, verbose: bool) -> bool:
|
|
pass
|
|
|
|
async def reboot_required(self, verbose: bool|None=None) -> bool:
|
|
if verbose is None:
|
|
verbose = self.ctx.verbose_default
|
|
return await self._reboot_required(verbose=verbose)
|
|
|
|
# -- select
|
|
|
|
@abc.abstractmethod
|
|
async def _select(self, names: Iterable[str]) -> Iterable[Package]:
|
|
pass
|
|
|
|
async def select(self, names: Iterable[str] = []) -> Iterable[Package]:
|
|
return await self._select(names)
|
|
|
|
# -- install
|
|
|
|
@abc.abstractmethod
|
|
async def _install(self, names: Iterable[str], only_update: bool) -> None:
|
|
pass
|
|
|
|
async def install(self, names: Iterable[str], only_update: bool=False) -> None:
|
|
return await self._install(names, only_update=only_update)
|
|
|
|
# -- delete
|
|
|
|
@abc.abstractmethod
|
|
async def _delete(self, names: Iterable[str]) -> None:
|
|
pass
|
|
|
|
async def delete(self, names: Iterable[str]) -> None:
|
|
return await self._delete(names)
|
|
|
|
# -- pkg_files
|
|
|
|
@abc.abstractmethod
|
|
async def _pkg_files(self, name: str) -> Iterable[str]:
|
|
pass
|
|
|
|
async def pkg_files(self, name: str) -> Iterable[str]:
|
|
return await self._pkg_files(name)
|