janware/jw-pkg
Watch
1
Code Pull requests Activity Actions

lib.util.copy(): Add parents flag parameter

Browse source 
copy() lands a source below the destination root, taking at most the
source's basename, so there is no way to land it at its source path, as cp
--parents does.

Add a parents flag that requests just that.

Signed-off-by: Jan Lindemann <jan@janware.com>
Assisted-by: unsloth/Qwen3.8-27B-GGUF:Q4_K_M with pi.dev v1.1.0
This commit is contained in:
Jan Lindemann 2026-10-04 15:56:48 +02:00
commit 0ca539dc0f
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
1 changed files with 68 additions and 3 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -176,6 +176,7 @@ async def copy(
mode: int | None = None,
throw: bool = True,
recursive: bool = False,
parents: bool = False,
) -> Exception | str | list[str]:
# Destination address / addresses
@ -183,7 +184,7 @@ async def copy(
if not isinstance(src, (str, FileContext)):
ret = []
for uri in src: # TODO: Group identical netlocs into one CopyContext
rr = await copy(uri, dst, owner, group, mode, throw, recursive)
rr = await copy(uri, dst, owner, group, mode, throw, recursive, parents)
if isinstance(rr, Exception):
return rr
if isinstance(rr, list):
@ -225,10 +226,32 @@ async def copy(
async with CopyContext(src, dst) as ctx:
src_uri = src.uri if isinstance(src, FileContext) else Uri(src)
ret = '' if ctx.dst.chroot else ctx.dst.root
if ret:
if parents:
# -- cp --parents: the source lands at its source path
# below the destination root, each parent level
# typed by the source-side node it mirrors
parts = [p for p in src_uri.path.split('/') if p]
if parts:
src_path = '/'.join(parts)
base = ret
ret = f'{base}/{src_path}' if base else src_path
s_cur = ''
for part in parts[:-1]:
s_cur = f'{s_cur}/{part}' if s_cur else part
d_cur = f'{base}/{s_cur}' if base else s_cur
if await ctx.dst.is_dir(d_cur):
continue
level_ctx = FileContext.create(
f'{ctx.src.uri.scheme_plus_authority}/{s_cur}',
chroot = True,
)
async with level_ctx:
ftype = (await level_ctx.stat('')).ftype
await ctx.dst.mkdir(d_cur, ftype = ftype)
elif ret:
# Land the source's basename within existing destination directories
src_uri = src.uri if isinstance(src, FileContext) else Uri(src)
basename = os.path.basename(src_uri.path.rstrip('/'))
if basename and await ctx.dst.is_dir(ret):
ret += '/' + basename