lib.Package: Add and fill version field

Add a version field to the Package class and make query_packages() fill it.
It's very much distribution and package dependent what shape that version
has, so from Package's perspective, it's just a str-typed member.

Further improvements and fixes:

- Make query_packages() accept throw, defaulting to the previous behaviour,
  so a caller that wants to be prepared for a missing package / an empty
  result can pass throw = False.
- Make dpkg's _run() and run_dpkg_query() forward ec and the remaining
  keyword arguments on the sudo path, which dropped them before.

Add unit tests for the file type tree and its StatResult predicates.

Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.4.0
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-10-27 14:59:14 +01:00
commit afbdc2e4b0
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
3 changed files with 104 additions and 32 deletions

View file

@ -8,19 +8,25 @@ meta_tags = [
'packager',
'url',
'maintainer',
'version',
]
class Package:
"""A package as reported by a package manager. The version is the
full version string as the PM reports it, for instance
version-release for rpm and the single version for dpkg."""
name: str
vendor: str | None = None
packager: str | None = None
url: str | None = None
maintainer: str | None = None
version: str | None = None
@classmethod
def parse_spec_str(cls, spec: str, delimiter: str = '|') -> 'Package':
tags = spec.split(delimiter)
if len(tags) != 5:
if len(tags) != len(meta_tags):
raise ValueError(f'Invalid package spec string "{spec}"')
return cls(
name = tags[0],
@ -28,6 +34,7 @@ class Package:
packager = tags[2],
url = tags[3],
maintainer = tags[4],
version = tags[5],
)
@classmethod
@ -51,12 +58,14 @@ class Package:
packager: str | None = None,
url: str | None = None,
maintainer: str | None = None,
version: str | None = None,
):
self.name = name
self.vendor = vendor
self.packager = packager
self.url = url
self.maintainer = maintainer
self.version = version
@override
def __repr__(self) -> str:

View file

@ -17,11 +17,12 @@ def meta_map() -> dict[str, str]:
if _meta_map is None:
_meta_map = Package.order_tags(
{
'name': 'binary:Package',
'name': '${binary:Package}',
'vendor': None, # deb doesn't have vendor field
'packager': None, # -- packager --
'url': 'Homepage',
'maintainer': 'Maintainer',
'url': '${Homepage}',
'maintainer': '${Maintainer}',
'version': '${Version}',
}
)
return _meta_map
@ -33,7 +34,7 @@ async def _run(
**kwargs: Any,
) -> str:
return (
await run_sudo(cmd) if sudo else await
await run_sudo(cmd, ec = ec, **kwargs) if sudo else await
run_cmd(cmd, ec = ec, cmd_input = InputMode.NonInteractive, **kwargs)
).stdout_str
@ -50,20 +51,23 @@ async def run_dpkg( # export
async def run_dpkg_query( # export
args: list[str],
sudo: bool = False,
ec: ExecContext | None = None
ec: ExecContext | None = None,
**kwargs: Any,
) -> str:
cmd = ['/usr/bin/dpkg-query']
cmd.extend(args)
return await _run(cmd, sudo, ec)
return await _run(cmd, sudo, ec, **kwargs)
async def query_packages(names: Iterable[str] = [],
ec: ExecContext | None = None) -> Iterable[Package]:
fmt_str = (
'|'.join([(f'${{{tag}}}' if tag else '')
for tag in meta_map().values()]) + r'\n'
)
async def query_packages(
names: Iterable[str] = [],
ec: ExecContext | None = None,
throw: bool = True
) -> Iterable[Package]:
fmt_str = '|'.join(tag or '' for tag in meta_map().values()) + r'\n'
# dpkg-query -W -f='${binary:Package}|${Maintainer}| ... \n'
specs = await run_dpkg_query(['-W', '-f=' + fmt_str, *names], sudo = False, ec = ec)
specs = await run_dpkg_query(
['-W', '-f=' + fmt_str, *names], sudo = False, throw = throw, ec = ec
)
return cast('Iterable[Package]', Package.parse_specs_str(specs))
async def list_files(

View file

@ -19,11 +19,13 @@ def meta_map() -> dict[str, str]:
if _meta_map is None:
_meta_map = Package.order_tags(
{
'name': 'Name',
'vendor': 'Vendor',
'packager': 'Packager',
'url': 'URL',
'name': '%{Name}',
'vendor': '%{Vendor}',
'packager': '%{Packager}',
'url': '%{URL}',
'maintainer': None, # RPM doesn't have a maintainer field
# -- the full version identity as one column
'version': '%{VERSION}-%{RELEASE}',
}
)
return _meta_map
@ -46,17 +48,15 @@ async def run_rpm( # export
async def query_packages( # export
names: Collection[str] = [],
ec: ExecContext | None = None,
throw: bool = True,
) -> Iterable[Package]:
fmt_str = (
'|'.join([(f'%{{{tag}}}' if tag else '')
for tag in meta_map().values()]) + r'\n'
)
fmt_str = '|'.join(tag or '' for tag in meta_map().values()) + r'\n'
opts = ['-q', '--queryformat', fmt_str]
if not names:
opts.append('-a')
specs = await run_rpm(
[*opts, *names],
throw = True,
throw = throw,
sudo = False,
mode = InputMode.NonInteractive,
ec = ec