janware/jw-pkg
1
Code Pull requests Activity Actions

lib.util.copy(): Tolerate list[str] and FileContext args

Browse source 
Modify copy():
  - Allow the source argument src_uri to be a list of URIs, in which
    case copy should return a list of paths instead of one path
  - Change the dst_uri parameter to dst, signalling that it now also
    tolerates a FileContext
  - Use the CopyContext class to manage the source and target
    FileContext instances
Signed-off-by: Jan Lindemann <jan@janware.com>
This commit is contained in:
Jan Lindemann 2026-04-27 07:33:27 +02:00
commit 5837d10a1c
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
1 changed files with 23 additions and 23 deletions
Download patch file Download diff file Expand all files Collapse all files

46
src/python/jw/pkg/lib/util.py
View file

@ -101,29 +101,29 @@ 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: int|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 await dst.is_dir(path):
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 copy(src_uri: str|Iterable[str], dst: str|FileContext, owner: str|None=None, group: str|None=None, mode: int|None=None, throw=True) -> Exception|str|list[str]:
if not isinstance(src_uri, str):
ret: list[str] = []
for uri in src_uri: # TODO: Group identical netlocs into one CopyContext
rr = ret.append(await copy(uri, dst, owner, group, mode, throw))
if isinstance(rr, Exception):
return rr
return ret
from .CopyContext import CopyContext
async with CopyContext(src_uri, dst) as ctx:
try:
content = (await ctx.src.get(ctx.src.root, throw=True)).stdout
dst_path = ctx.dst.root
if await ctx.dst.is_dir(ctx.dst.root):
dst_path += '/' + os.path.basename(src_uri)
await ctx.dst.put(path=dst_path, content=content, owner=owner, group=group, mode=mode, throw=True)
return dst_path
except Exception as e:
if throw:
raise
log(ERR, f'Failed to copy {src_uri} -> {dst} ({str(e)})')
return e
assert False, 'Unreachable code'
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 Uri(url).username