diff --git a/src/python/jw/pkg/lib/distros/redhat/Distro.py b/src/python/jw/pkg/lib/distros/redhat/Distro.py new file mode 100644 index 00000000..0ba5d473 --- /dev/null +++ b/src/python/jw/pkg/lib/distros/redhat/Distro.py @@ -0,0 +1,79 @@ +from __future__ import annotations + +from typing import TYPE_CHECKING, override + +from ...Distro import Distro as Base +from ...pm.rpm import list_files, query_packages, run_rpm + +if TYPE_CHECKING: + from typing import Any, Collection, Iterable + + from ...base import Result + from ...ExecContext import ExecContext + from ...Package import Package + +class Distro(Base): + + async def yum( + self, args: list[str], verbose: bool = True, sudo: bool = True + ) -> Result: + cmd = ['/usr/bin/yum'] + cmd.extend(args) + if sudo: + return await self.sudo(cmd, verbose = verbose) + return await self.run(cmd, verbose = verbose) + + async def rpm( + self, + *args: Any, + ec: ExecContext | None = None, + **kwargs: Any, + ) -> str: + if ec is None: + ec = self.ctx + kwargs['ec'] = ec + return await run_rpm(*args, **kwargs) + + def __init__(self, *args: Any, **kwargs: Any) -> None: + super().__init__(*args, **kwargs) + + @override + async def _ref(self) -> None: + await self.yum(['clean', 'expire-cache']) + await self.yum(['makecache']) + + @override + async def _dup(self, download_only: bool) -> None: + args: list[str] = ['update'] + if download_only: + args.append('--downloadonly') + await self.yum(args) + + @override + async def _reboot_required(self, verbose: bool) -> bool: + raise NotImplementedError() + + @override + async def _select_by_name(self, names: Collection[str]) -> Iterable[Package]: + return await query_packages(names, ec = self.ctx) + + @override + async def _install(self, names: Iterable[str], only_update: bool) -> None: + args = ['update' if only_update else 'install'] + if not self.interactive: + args.append('-y') + args.extend(names) + await self.yum(args) + + @override + async def _delete(self, names: Iterable[str]) -> None: + await self.rpm(['-e', *names], sudo = True) + + @override + async def _pkg_files(self, name: str) -> Iterable[str]: + return await list_files(name, ec = self.ctx) + + async def _download(self, name: str, tmp_dir: str) -> str: + raise NotImplementedError( + 'distro pkg download is not yet implemented for RedHat distributions' + )