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

# -*- coding: utf-8 -*-

from __future__ import annotations

from typing import TYPE_CHECKING

if TYPE_CHECKING:
    import Iterable

import abc, importlib

from .ExecContext import ExecContext
from .base import 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
        self.__id: str|None = None

    def __set_id(self, id: str) -> None:
        self.__id = id

    # == 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')
        ret = cls(*args, **kwargs)
        ret.__set_id(backend_id)
        return ret

    # == Convenience methods

    @property
    def id(self) -> str:
        return self.__id

    @property
    def ctx(self) -> ExecContext:
        return self.__exec_context

    async def run(self, *args, **kwargs) -> Result:
        return await self.__exec_context.run(*args, **kwargs)

    async def sudo(self, *args, **kwargs) -> Result:
        return await 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

    # Pass names to the package manager
    @abc.abstractmethod
    async def _install(self, names: Iterable[str], only_update: bool) -> None:
        pass

    # Default implementation assumes package manager can handle local files.
    # Not true for all distros. Override if Distro knows better.
    async def _install_local_files(self, paths: Iterable[str], only_update: bool) -> None:
        await self._install(paths, only_update=only_update)

    # Download first and then install. Override if Distro knows better.
    async def _install_urls(self, urls: Iterable[str], only_update: bool) -> None:
        import tempfile
        from .util import copy
        with tempfile.TemporaryDirectory(prefix='jw-pkg-') as tmp:
            paths: list[str] = []
            for url in urls:
                paths.append(await copy(url, tmp))
            await self._install_local_files(paths, only_update=only_update)

    # Default implementation installs in two steps:
    # - Download URLs into local directories and install
    # - Pass names to package manager
    # Override if Distro knows better.
    async def _install_urls_and_names(self, packages: Iterable[str], only_update: bool) -> None:
        urls: list[str] = []
        names: list[str] = []
        for package in packages:
            if package[0] == '/':
                urls.append('file://' + package)
                continue
            if package.find('://') != -1:
                urls.append(package)
                continue
            names.append(package)
        if urls:
            await self._install_urls(urls, only_update=only_update)
        if names:
            await self._install(names, only_update=only_update)

    async def install(self, names: Iterable[str], only_update: bool=False) -> None:
        if not names:
            log(WARNING, f'No packages specified for installation')
            return
        await self._install_urls_and_names(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:
        if not names:
            log(WARNING, f'No packages specified for deletion')
            return
        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]:
        if not name:
            log(WARNING, f'No package specified for inspection')
            return []
        return await self._pkg_files(name)