from __future__ import annotations from typing import TYPE_CHECKING, Iterable, cast if TYPE_CHECKING: from ..ExecContext import ExecContext from ..base import InputMode from ..Package import Package from ..util import run_cmd, run_sudo _meta_map: dict[str, str] | None = None def meta_map() -> dict[str, str]: global _meta_map if _meta_map is None: _meta_map = Package.order_tags( { 'name': 'binary:Package', 'vendor': None, # deb doesn't have vendor field 'packager': None, # -- packager -- 'url': 'Homepage', 'maintainer': 'Maintainer', } ) return _meta_map async def _run( cmd: list[str], sudo: bool = False, ec: ExecContext | None = None ) -> str: return ( await run_sudo(cmd, ec = ec, cmd_input = InputMode.NonInteractive) if sudo else await run_cmd(cmd, ec = ec, cmd_input = InputMode.NonInteractive) ).stdout_str async def run_dpkg( # export args: list[str], sudo: bool = False, ec: ExecContext | None = None ) -> str: cmd = ['/usr/bin/dpkg'] cmd.extend(args) return await _run(cmd, sudo, ec) async def run_dpkg_query( # export args: list[str], sudo: bool = False, ec: ExecContext | None = None ) -> str: cmd = ['/usr/bin/dpkg-query'] cmd.extend(args) return await _run(cmd, sudo, ec) 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' ) # dpkg-query -W -f='${binary:Package}|${Maintainer}| ... \n' specs = await run_dpkg_query(['-W', '-f=' + fmt_str, *names], sudo = False, ec = ec) return cast('Iterable[Package]', Package.parse_specs_str(specs)) async def list_files(pkg: str, ec: ExecContext | None = None) -> list[str]: file_list_str = await run_dpkg(['-L', pkg], sudo = False, ec = ec) return file_list_str.splitlines()