mirror of
ssh://git.janware.com/janware/proj/jw-pkg
synced 2026-04-24 09:13:37 +02:00
Jan Lindemann
3e897f4df8
The code below lib.distro, as left behind by the previous commit, is geared towards being directly used as a command-line API. This commit introduces the abstract base class Distro, a proxy for distribution-specific interactions. The proxy abstracts distro specifics into an API with proper method prototypes, not argparse.Namespace contents, and can thus be more easily driven by arbitrary code. The Distro class is initialized with a member variable of type ExecContext, another new class introduced by this commit. It is designed to abstract the communication channel to the distribution instance. Currently only one specialization exists, Local, which interacts with the distribution and root file system it is running in, but is planned to be subclassed to support interaction via SSH, serial, chroot, or chains thereof. Signed-off-by: Jan Lindemann <jan@janware.com>
75 lines
2.4 KiB
Python
75 lines
2.4 KiB
Python
# -*- coding: utf-8 -*-
|
|
|
|
from __future__ import annotations
|
|
from typing import TYPE_CHECKING
|
|
|
|
import os
|
|
|
|
from ...log import *
|
|
from ...Distro import Distro as Base
|
|
from ...pm.dpkg import run_dpkg, run_dpkg_query, query_packages, list_files
|
|
|
|
if TYPE_CHECKING:
|
|
from typing import Iterable
|
|
from ...ExecContext import Result
|
|
from ...Package import Package
|
|
|
|
class Distro(Base):
|
|
|
|
async def apt_get(self, args: list[str]):
|
|
cmd = ['/usr/bin/apt-get']
|
|
mod_env = None
|
|
if not self.interactive:
|
|
cmd.extend(['--yes', '--quiet'])
|
|
mod_env = { 'DEBIAN_FRONTEND': 'noninteractive' }
|
|
cmd.extend(args)
|
|
return await self.sudo(cmd, mod_env=mod_env)
|
|
|
|
async def dpkg(self, *args, **kwargs):
|
|
return await run_dpkg(*args, **kwargs)
|
|
|
|
def __init__(self, *args, **kwargs):
|
|
super().__init__(*args, **kwargs)
|
|
|
|
async def _ref(self) -> None:
|
|
return await self.apt_get(['update'])
|
|
|
|
async def _dup(self, download_only: bool) -> None:
|
|
args: list[str] = []
|
|
if download_only:
|
|
args.append('--download-only')
|
|
args.append('upgrade')
|
|
return await self.apt_get(args)
|
|
|
|
async def _reboot_required(self, verbose: bool) -> bool:
|
|
reboot_required = '/run/reboot_required'
|
|
if os.path.exists(reboot_required):
|
|
if verbose:
|
|
log(NOTICE, f'Yes. {reboot_required} exists.')
|
|
required_pkgs = '/run/reboot-required.pkgs'
|
|
if os.path.exists(required_pkgs):
|
|
with open(required_pkgs, 'r') as f:
|
|
content = f.read()
|
|
print(f'-- From {required_pkgs}:')
|
|
print(content.strip())
|
|
return True
|
|
if verbose:
|
|
log(NOTICE, f'No. {reboot_required} doesn\'t exist.')
|
|
return False
|
|
|
|
async def _select(self, names: Iterable[str]) -> Iterable[Package]:
|
|
return await query_packages(names)
|
|
|
|
async def _install(self, names: Iterable[str], only_update: bool) -> None:
|
|
args = ['install']
|
|
if only_update:
|
|
args.append('--only-upgrade')
|
|
args.append('--no-install-recommends')
|
|
args.extend(names)
|
|
return await self.apt_get(args)
|
|
|
|
async def _delete(self, names: Iterable[str]) -> None:
|
|
return await self.dpkg(['-P', *names], sudo=True)
|
|
|
|
async def _pkg_files(self, name: str) -> Iterable[str]:
|
|
return await list_files(name)
|