From 63383cb6837410932a802e189612a2b7affc5fda Mon Sep 17 00:00:00 2001 From: Jan Lindemann Date: Wed, 15 Apr 2026 14:06:35 +0200 Subject: [PATCH] lib.util.copy(): Add function Add copy(src_uri, dst_uri), instatiating two ExecContext instances, and doing the obvious with them - copying from src_uri to dst_uri. Signed-off-by: Jan Lindemann --- src/python/jw/pkg/lib/util.py | 28 +++++++++++++++++++++++++++- 1 file changed, 27 insertions(+), 1 deletion(-) diff --git a/src/python/jw/pkg/lib/util.py b/src/python/jw/pkg/lib/util.py index 04537527..ec74e7cb 100644 --- a/src/python/jw/pkg/lib/util.py +++ b/src/python/jw/pkg/lib/util.py @@ -6,7 +6,7 @@ from typing import TYPE_CHECKING, Iterable if TYPE_CHECKING: from typing import Sequence - from ExecContext import ExecContext + from .ExecContext import ExecContext import os, sys, json @@ -99,6 +99,32 @@ async def run_sudo(cmd: list[str], *args, interactive: bool=True, ec: ExecContex ec = Local(interactive=interactive) return await ec.sudo(cmd, *args, **kwargs) +async def copy(src_uri: str, dst_uri: str, owner: str|None=None, group: str|None=None, mode: str|None=None, throw=True) -> Exception|str: + from .ExecContext import ExecContext + src: ExecContext|None = None + dst: ExecContext|None = None + def __ec(uri: str) -> tuple[ExecContext, str]: + return ExecContext.create(uri), urlparse(uri).path + try: + src, src_path = __ec(src_uri) + content = (await src.get(src_path, throw=True)).stdout + dst, dst_path = __ec(dst_uri) + if os.path.isdir(dst_path) and not dst_path[-1] == '/': + dst_path += '/' + if dst_path[-1] == '/': + dst_path += os.path.basename(src_path) + await dst.put(path=dst_path, content=content, owner=owner, group=group, mode=mode, throw=True) + except Exception as e: + if throw: + raise + log(ERR, f'Failed to copy {src_uri} -> {dst_uri} ({str(e)})') + return e + finally: + for ec in [src, dst]: + if ec is not None: + await ec.close() + return dst_path + async def get_username(args: Namespace|None=None, url: str|None=None, askpass_env: list[str]=[], ec: ExecContext|None=None) -> str: # export url_user = None if url is None else urlparse(url).username if args is not None: