jw-pkg/src/python/jw/pkg/lib/pm/dpkg.py

# -*- coding: utf-8 -*-

from __future__ import annotations

from typing import Iterable, TYPE_CHECKING

if TYPE_CHECKING:
    from ..ExecContext import ExecContext

from ..ExecContext import InputMode
from ..util import run_cmd, run_sudo
from ..Package import Package, meta_tags

_meta_map: dict[str, str]|None = None

def meta_map():
    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_dpkg(args: list[str], sudo: bool=False, ec: ExecContext=None): # export
    cmd = ['/usr/bin/dpkg']
    cmd.extend(args)
    if sudo:
        return await run_sudo(cmd, ec=ec)
    return (await run_cmd(cmd, ec=ec)).decode()

async def run_dpkg_query(args: list[str], sudo: bool=False, ec: ExecContext=None): # export
    cmd = ['/usr/bin/dpkg-query']
    cmd.extend(args)
    if sudo:
        return await run_sudo(cmd)
    return (await run_cmd(cmd, ec=ec, cmd_input=InputMode.NonInteractive)).decode()

async def query_packages(names: Iterable[str] = [], ec: ExecContext=None) -> Iterable[Package]: # export
    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, stderr, status  = await run_dpkg_query(['-W', '-f=' + fmt_str, *names], sudo=False, ec=ec)
    return Package.parse_specs_str(specs)

async def list_files(pkg: str, ec: ExecContext=None) -> list[str]: # export
    file_list_str, stderr, status  = await run_dpkg(['-L', pkg], sudo=False, ec=ec)
    return file_list_str.splitlines()