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'
+        )