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

lib.util.copy(): Accept a FileContext as source

Browse source 
Using a FileContext, notably with its chroot flag set, is useful as a
tree source in copy(), so allow it.

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 13:55:39 +02:00
commit 482daa17e7
Signed by: Jan Lindemann
GPG key ID: 3750640C9E25DD61
1 changed files with 32 additions and 7 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -8,6 +8,7 @@ from enum import Enum, auto
from typing import TYPE_CHECKING, Any, Iterable, TypeVar, cast
from .base import Input, InputMode, Result
from .FileContext import FileContext
from .log import DEBUG, ERR, log
from .Uri import Uri
@ -15,7 +16,6 @@ if TYPE_CHECKING:
from argparse import Namespace
from .ExecContext import ExecContext
from .FileContext import FileContext
from .ProcFilter import ProcFilter, ProcPipeline
T = TypeVar('T')
@ -169,7 +169,7 @@ async def get(
return await run_pipeline(await ctx.get(uri.path, *args, **kwargs), content_filter)
async def copy(
src_uri: str | Iterable[str],
src: str | FileContext | Iterable[str],
dst: str | FileContext,
owner: str | None = None,
group: str | None = None,
@ -180,9 +180,9 @@ async def copy(
# Destination address / addresses
ret: str | list[str]
if not isinstance(src_uri, str):
if not isinstance(src, (str, FileContext)):
ret = []
for uri in src_uri: # TODO: Group identical netlocs into one CopyContext
for uri in src: # TODO: Group identical netlocs into one CopyContext
rr = await copy(uri, dst, owner, group, mode, throw, recursive)
if isinstance(rr, Exception):
return rr
@ -223,12 +223,13 @@ async def copy(
from .CopyContext import CopyContext
async with CopyContext(src_uri, dst) as ctx:
async with CopyContext(src, dst) as ctx:
ret = '' if ctx.dst.chroot else ctx.dst.root
if ret:
# Land the source's basename within existing destination directories
basename = os.path.basename(Uri(src_uri).path.rstrip('/'))
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
src_root = '' if ctx.src.chroot else ctx.src.root
@ -251,7 +252,7 @@ async def copy(
except Exception as e:
if throw:
raise
log(ERR, f'Failed to copy {src_uri} -> {dst} ({str(e)})')
log(ERR, f'Failed to copy {src} -> {dst} ({str(e)})')
return e
async def get_username( # export