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 <jan@janware.com>

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()

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_sudo(cmd, **kwargs)
-    return await run_cmd(cmd)
+    return await run_cmd(cmd, **kwargs)
 
-async def all_installed_packages() -> Iterable[Package]: # export
+async def query_packages(names: Iterable[str] = []) -> Iterable[Package]: # export
-    ret: list[Package] = []
+    fmt_str = '|'.join([(f'%{{{tag}}}' if tag else '') for tag in meta_map().values()]) + r'\n'
-    opts: list[str] = []
+    opts = ['-q', '--queryformat', fmt_str]
-    query_tags: list[str] = []
+    if not names:
-    query_tags.append('Name')
+        opts.append('-a')
-    query_tags.append('Vendor')
+    specs, stderr, status  = await run_rpm([*opts, *names], throw=True, sudo=False)
-    query_tags.append('Packager')
+    return Package.parse_specs_str(specs)
-    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 list_files(pkg: str) -> list[str]: # export
-    opts: list[str] = []
+    file_list_str, stderr, status  = await run_rpm(['-ql', pkg], throw=True, sudo=False)
-    file_list_str, stderr, status  = await run_rpm(['-ql', pkg, *opts], sudo=False)
     return file_list_str.splitlines()