janware/jw-pkg
1
mirror of ssh://git.janware.com/janware/proj/jw-pkg synced 2026-04-24 17:23:36 +02:00
Code Activity Actions

lib.pm: Add parameter ec: ExecContext to functions

Browse source 
Functions in lib.pm (i.e. run_dpkg(), run_rpm() and friends) also get
an ExecContext-type parameter. Use them in lib/distros/*/Distro.

Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-03-06 17:25:19 +01:00
commit 525fa34387
5 changed files with 37 additions and 27 deletions
Download patch file Download diff file Expand all files Collapse all files

6
src/python/jw/pkg/lib/distros/arch/Distro.py
View file

@ -29,20 +29,20 @@ class Distro(Base):
args = ['-Su'] args = ['-Su']
if args.download_only: if args.download_only:
args.append('-w') args.append('-w')
return await self.util.pacman(args) return await self.pacman(args)
async def _reboot_required(self, verbose: bool) -> bool: async def _reboot_required(self, verbose: bool) -> bool:
raise NotImplementedError('distro reboot-required is not yet implemented for Arch-like distributions') raise NotImplementedError('distro reboot-required is not yet implemented for Arch-like distributions')
async def _select(self, names: Iterable[str]) -> Iterable[Package]: async def _select(self, names: Iterable[str]) -> Iterable[Package]:
return await query_packages(names) raise NotImplementedError('distro select is not yet implemented for Arch-like distributions')
async def _install(self, names: Iterable[str], only_update: bool) -> None: async def _install(self, names: Iterable[str], only_update: bool) -> None:
if only_update: if only_update:
raise NotImplementedError('--only-update is not yet implemented for pacman') raise NotImplementedError('--only-update is not yet implemented for pacman')
args = ['-S', '--needed'] args = ['-S', '--needed']
args.extend(args.packages) args.extend(args.packages)
await self.util.pacman(args) await self.pacman(args)
async def _delete(self, names: Iterable[str]) -> None: async def _delete(self, names: Iterable[str]) -> None:
raise NotImplementedError('distro delete not yet implemented for Arch-like distributions') raise NotImplementedError('distro delete not yet implemented for Arch-like distributions')

6
src/python/jw/pkg/lib/distros/debian/Distro.py
View file

@ -26,7 +26,7 @@ class Distro(Base):
return await self.sudo(cmd, mod_env=mod_env) return await self.sudo(cmd, mod_env=mod_env)
async def dpkg(self, *args, **kwargs): async def dpkg(self, *args, **kwargs):
return await run_dpkg(*args, **kwargs) return await run_dpkg(*args, ec=self.ctx, **kwargs)
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
@ -58,7 +58,7 @@ class Distro(Base):
return False return False
async def _select(self, names: Iterable[str]) -> Iterable[Package]: async def _select(self, names: Iterable[str]) -> Iterable[Package]:
return await query_packages(names) return await query_packages(names, ec=self.ctx)
async def _install(self, names: Iterable[str], only_update: bool) -> None: async def _install(self, names: Iterable[str], only_update: bool) -> None:
args = ['install'] args = ['install']
@ -72,4 +72,4 @@ class Distro(Base):
return await self.dpkg(['-P', *names], sudo=True) return await self.dpkg(['-P', *names], sudo=True)
async def _pkg_files(self, name: str) -> Iterable[str]: async def _pkg_files(self, name: str) -> Iterable[str]:
return await list_files(name) return await list_files(name, ec=self.ctx)

6
src/python/jw/pkg/lib/distros/suse/Distro.py
View file

@ -24,7 +24,7 @@ class Distro(Base):
return await self.run(cmd, verbose=verbose) return await self.run(cmd, verbose=verbose)
async def rpm(self, *args, **kwargs) -> Result: async def rpm(self, *args, **kwargs) -> Result:
return await run_rpm(*args, **kwargs) return await run_rpm(*args, ec=self.ctx, **kwargs)
def __init__(self, *args, **kwargs): def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs) super().__init__(*args, **kwargs)
@ -50,7 +50,7 @@ class Distro(Base):
return False return False
async def _select(self, names: Iterable[str]) -> Iterable[Package]: async def _select(self, names: Iterable[str]) -> Iterable[Package]:
return await query_packages(names) return await query_packages(names, ec=self.ctx)
async def _install(self, names: Iterable[str], only_update: bool) -> None: async def _install(self, names: Iterable[str], only_update: bool) -> None:
cmd = 'update' if only_update else 'install' cmd = 'update' if only_update else 'install'
@ -60,4 +60,4 @@ class Distro(Base):
return await self.rpm(['-e', *names], sudo=True) return await self.rpm(['-e', *names], sudo=True)
async def _pkg_files(self, name: str) -> Iterable[str]: async def _pkg_files(self, name: str) -> Iterable[str]:
return await list_files(name) return await list_files(name, ec=self.ctx)

25
src/python/jw/pkg/lib/pm/dpkg.py
View file

@ -1,6 +1,11 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
from typing import Iterable from __future__ import annotations
from typing import Iterable, TYPE_CHECKING
if TYPE_CHECKING:
from ..ExecContext import ExecContext
from ..util import run_cmd, run_sudo from ..util import run_cmd, run_sudo
@ -20,26 +25,26 @@ def meta_map():
}) })
return _meta_map return _meta_map
async def run_dpkg(args: list[str], sudo: bool=False): # export async def run_dpkg(args: list[str], sudo: bool=False, ec: ExecContext=None): # export
cmd = ['/usr/bin/dpkg'] cmd = ['/usr/bin/dpkg']
cmd.extend(args) cmd.extend(args)
if sudo: if sudo:
return await run_sudo(cmd) return await run_sudo(cmd, ec=ec)
return await run_cmd(cmd) return await run_cmd(cmd, ec=ec)
async def run_dpkg_query(args: list[str], sudo: bool=False): # export async def run_dpkg_query(args: list[str], sudo: bool=False, ec: ExecContext=None): # export
cmd = ['/usr/bin/dpkg-query'] cmd = ['/usr/bin/dpkg-query']
cmd.extend(args) cmd.extend(args)
if sudo: if sudo:
return await run_sudo(cmd) return await run_sudo(cmd)
return await run_cmd(cmd) return await run_cmd(cmd, ec=ec)
async def query_packages(names: Iterable[str] = []) -> Iterable[Package]: # export 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' fmt_str = '|'.join([(f'${{{tag}}}' if tag else '') for tag in meta_map().values()]) + r'\n'
# dpkg-query -W -f='${binary:Package}|${Maintainer}| ... \n' # dpkg-query -W -f='${binary:Package}|${Maintainer}| ... \n'
specs, stderr, status = await run_dpkg_query(['-W', '-f=' + fmt_str, *names], sudo=False) specs, stderr, status = await run_dpkg_query(['-W', '-f=' + fmt_str, *names], sudo=False, ec=ec)
return Package.parse_specs_str(specs) return Package.parse_specs_str(specs)
async def list_files(pkg: str) -> list[str]: # export async def list_files(pkg: str, ec: ExecContext=None) -> list[str]: # export
file_list_str, stderr, status = await run_dpkg(['-L', pkg], sudo=False) file_list_str, stderr, status = await run_dpkg(['-L', pkg], sudo=False, ec=ec)
return file_list_str.splitlines() return file_list_str.splitlines()

21
src/python/jw/pkg/lib/pm/rpm.py
View file

@ -1,6 +1,11 @@
# -*- coding: utf-8 -*- # -*- coding: utf-8 -*-
from typing import Iterable from __future__ import annotations
from typing import Iterable, TYPE_CHECKING
if TYPE_CHECKING:
from ..ExecContext import ExecContext
from ..util import run_cmd, run_sudo from ..util import run_cmd, run_sudo
@ -20,21 +25,21 @@ def meta_map():
}) })
return _meta_map return _meta_map
async def run_rpm(args: list[str], sudo: bool=False, **kwargs): # export async def run_rpm(args: list[str], sudo: bool=False, ec: ExecContext=None, **kwargs): # export
cmd = ['/usr/bin/rpm'] cmd = ['/usr/bin/rpm']
cmd.extend(args) cmd.extend(args)
if sudo: if sudo:
return await run_sudo(cmd, **kwargs) return await run_sudo(cmd, ec=ec, **kwargs)
return await run_cmd(cmd, **kwargs) return await run_cmd(cmd, ec=ec, **kwargs)
async def query_packages(names: Iterable[str] = []) -> Iterable[Package]: # export 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' fmt_str = '|'.join([(f'%{{{tag}}}' if tag else '') for tag in meta_map().values()]) + r'\n'
opts = ['-q', '--queryformat', fmt_str] opts = ['-q', '--queryformat', fmt_str]
if not names: if not names:
opts.append('-a') opts.append('-a')
specs, stderr, status = await run_rpm([*opts, *names], throw=True, sudo=False) specs, stderr, status = await run_rpm([*opts, *names], throw=True, sudo=False, ec=ec)
return Package.parse_specs_str(specs) return Package.parse_specs_str(specs)
async def list_files(pkg: str) -> list[str]: # export async def list_files(pkg: str, ec: ExecContext=None) -> list[str]: # export
file_list_str, stderr, status = await run_rpm(['-ql', pkg], throw=True, sudo=False) file_list_str, stderr, status = await run_rpm(['-ql', pkg], throw=True, sudo=False, ec=ec)
return file_list_str.splitlines() return file_list_str.splitlines()