jw-pkg/src/python/jw/pkg/lib/distros/debian/Distro.py

# -*- 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 ...base import Result
    from ...Package import Package

class Distro(Base):

    async def apt_get(self, args: list[str], verbose: bool=True, sudo: bool=True):
        cmd = ['/usr/bin/apt-get']
        mod_env = None
        if not self.interactive:
            cmd.extend(['--yes', '--quiet'])
            mod_env = { 'DEBIAN_FRONTEND': 'noninteractive' }
        cmd.extend(args)
        if sudo:
            return await self.sudo(cmd, verbose=verbose, mod_env=mod_env)
        return await self.run(cmd, verbose=verbose)

    async def dpkg(self, *args, **kwargs):
        return await run_dpkg(*args, ec=self.ctx, **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, ec=self.ctx)

    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, ec=self.ctx)