From 1bb3c166d60fd9a81901969f045b79ee1e98b31c Mon Sep 17 00:00:00 2001 From: Jan Lindemann Date: Wed, 4 Mar 2026 09:52:44 +0100 Subject: [PATCH] cmds.distro.lib.rpm.query_packages(): Add function Replace all_installed_packages() by query_packages(). The function takes an optional list of packages to be queried. If it's empty, a list of all installed packages are returned. Signed-off-by: Jan Lindemann --- .../jw/pkg/cmds/distro/backend/suse/Select.py | 4 +- src/python/jw/pkg/cmds/distro/lib/rpm.py | 48 ++++++++++--------- 2 files changed, 28 insertions(+), 24 deletions(-) diff --git a/src/python/jw/pkg/cmds/distro/backend/suse/Select.py b/src/python/jw/pkg/cmds/distro/backend/suse/Select.py index bafe0dba..913f48b9 100644 --- a/src/python/jw/pkg/cmds/distro/backend/suse/Select.py +++ b/src/python/jw/pkg/cmds/distro/backend/suse/Select.py @@ -4,7 +4,7 @@ from typing import Iterable from argparse import Namespace from ...lib.Package import Package -from ...lib.rpm import all_installed_packages +from ...lib.rpm import query_packages from ...Cmd import Cmd from ..BeSelect import BeSelect as Base @@ -14,4 +14,4 @@ class Select(Base): super().__init__(parent) async def _all_installed_packages(self) -> Iterable[Package]: - return await all_installed_packages() + return await query_packages() diff --git a/src/python/jw/pkg/cmds/distro/lib/rpm.py b/src/python/jw/pkg/cmds/distro/lib/rpm.py index 509cc15f..2edda5c6 100644 --- a/src/python/jw/pkg/cmds/distro/lib/rpm.py +++ b/src/python/jw/pkg/cmds/distro/lib/rpm.py @@ -4,33 +4,37 @@ from typing import Iterable from ....lib.util import run_cmd, run_sudo -from .Package import Package +from .Package import Package, meta_tags -async def run_rpm(args: list[str], sudo: bool=False): # export +_meta_map: dict[str, str]|None = None + +def meta_map(): + global _meta_map + if _meta_map is None: + _meta_map = Package.order_tags({ + 'name': 'Name', + 'vendor': 'Vendor', + 'packager': 'Packager', + 'url': 'URL', + 'maintainer': None, # RPM doesn't have a maintainer field + }) + return _meta_map + +async def run_rpm(args: list[str], sudo: bool=False, **kwargs): # export cmd = ['/usr/bin/rpm'] cmd.extend(args) if sudo: - return await run_sudo(cmd) - return await run_cmd(cmd) + return await run_sudo(cmd, **kwargs) + return await run_cmd(cmd, **kwargs) -async def all_installed_packages() -> Iterable[Package]: # export - ret: list[Package] = [] - opts: list[str] = [] - query_tags: list[str] = [] - query_tags.append('Name') - query_tags.append('Vendor') - query_tags.append('Packager') - query_tags.append('URL') - opts.append('--queryformat') - tag_str = '|'.join([f'%{{{tag}}}' for tag in query_tags]) + r'\n' - opts.append(tag_str) - package_list_str, stderr, status = await run_rpm(['-qa', *opts], sudo=False) - for package_str in package_list_str.splitlines(): - tags = package_str.split('|') - ret.append(Package(name=tags[0], vendor=tags[1], packager=tags[2], url=tags[3])) - return ret +async def query_packages(names: Iterable[str] = []) -> Iterable[Package]: # export + fmt_str = '|'.join([(f'%{{{tag}}}' if tag else '') for tag in meta_map().values()]) + r'\n' + opts = ['-q', '--queryformat', fmt_str] + if not names: + opts.append('-a') + specs, stderr, status = await run_rpm([*opts, *names], throw=True, sudo=False) + return Package.parse_specs_str(specs) async def list_files(pkg: str) -> list[str]: # export - opts: list[str] = [] - file_list_str, stderr, status = await run_rpm(['-ql', pkg, *opts], sudo=False) + file_list_str, stderr, status = await run_rpm(['-ql', pkg], throw=True, sudo=False) return file_list_str.splitlines()