Jan Lindemann
5d77955ed9
If /usr/bin/isort is found, run it during "make format" to get a defined way the imports are sorted. tool.isort in pyproject.toml is updated to match the other fixers. Commit the fallout of this change. Running the other fixers alone doesn't change the formatting, so this should be safe. Signed-off-by: Jan Lindemann <jan@janware.com>
77 lines
2.4 KiB
Python
77 lines
2.4 KiB
Python
from __future__ import annotations
|
|
|
|
import io
|
|
import tarfile
|
|
|
|
from tarfile import TarFile
|
|
from typing import TYPE_CHECKING, Callable
|
|
|
|
from ....lib.ExecContext import ExecContext
|
|
from ....lib.log import DEBUG, log
|
|
|
|
if TYPE_CHECKING:
|
|
from typing import Iterable
|
|
|
|
from ....lib.FileContext import FileContext
|
|
|
|
def filter(
|
|
blob: bytes,
|
|
path_filter: Callable[[str], bool] | None,
|
|
matched: list[str] | None = None,
|
|
) -> bytes:
|
|
ret = io.BytesIO()
|
|
with tarfile.open(fileobj = ret, mode = 'w') as tf_out:
|
|
tf_in = TarFile(fileobj = io.BytesIO(blob))
|
|
for info in tf_in.getmembers():
|
|
if path_filter is not None and not path_filter(info.name):
|
|
continue
|
|
log(DEBUG, f'Adding {info.name}')
|
|
if matched is not None:
|
|
matched.append(info.name)
|
|
buf = tf_in.extractfile(info)
|
|
tf_out.addfile(info, buf)
|
|
return ret.getvalue()
|
|
|
|
def rewrite(blob: bytes, rewrite_filter: Callable[[str], str]) -> bytes:
|
|
ret = io.BytesIO()
|
|
with tarfile.open(fileobj = ret, mode = 'w') as tf_out:
|
|
tf_in = TarFile(fileobj = io.BytesIO(blob))
|
|
for info in tf_in.getmembers():
|
|
new_name = rewrite_filter(info.name)
|
|
log(DEBUG, f'Rewriting {info.name} -> {new_name}')
|
|
info.name = new_name
|
|
buf = tf_in.extractfile(info)
|
|
tf_out.addfile(info, buf)
|
|
return ret.getvalue()
|
|
|
|
def merge(blobs: Iterable[bytes], overwrite: bool = False) -> bytes:
|
|
ret = io.BytesIO()
|
|
with tarfile.open(fileobj = ret, mode = 'w') as tf_out:
|
|
for blob in blobs:
|
|
tf_in = TarFile(fileobj = io.BytesIO(blob))
|
|
existing_names = tf_out.getnames()
|
|
for info in tf_in.getmembers():
|
|
if not overwrite and info.name in existing_names:
|
|
continue
|
|
buf = tf_in.extractfile(info)
|
|
tf_out.addfile(info, buf)
|
|
return ret.getvalue()
|
|
|
|
async def extract(
|
|
dst: FileContext,
|
|
blob: bytes,
|
|
root: str | None = None,
|
|
verbose: bool = False
|
|
) -> None:
|
|
cmd = ['tar']
|
|
if root is not None:
|
|
cmd += ['-C', root]
|
|
if verbose:
|
|
cmd += '-v'
|
|
cmd += ['-x', '-f', '-']
|
|
if not isinstance(dst, ExecContext):
|
|
raise NotImplementedError(
|
|
'Extracting tar files to a non-executable '
|
|
f'context is not yet implemented: {dst}'
|
|
)
|
|
await dst.run(cmd, verbose = verbose, cmd_input = blob)
|