WIP: keyring init

utils.py

@@ -138,6 +138,42 @@ def read_files_from_tar(tar_file: str, files: Sequence[str]) -> Generator[tuple[
             yield path, fd
 
 
+def read_files_from_tar_recursive(tar_file: str, paths: Sequence[str], append_slash: bool = True) -> Generator[tuple[str, IO], None, None]:
+    """
+    Returns tar FDs to files that lie under the directories specified in paths.
+    HINT: deactivate append_slash to get glob-like behaviour, as if all paths ended with *
+    """
+    assert os.path.exists(tar_file)
+    paths = [f"{p.strip('/')}/" for p in paths]
+    with tarfile.open(tar_file) as index:
+        for member in index.getmembers():
+            for path in paths:
+                if member.isfile() and member.path.startswith(path):
+                    fd = index.extractfile(member)
+                    assert fd
+                    yield member.path, fd
+                    break
+            continue
+
+
+def extract_files_from_tar_generator(
+    tar_generator: Generator[tuple[str, IO], None, None],
+    output_dir: str,
+    remove_prefix: str = '',
+    append_slash: bool = True,
+):
+    assert os.path.exists(output_dir)
+    remove_prefix = remove_prefix.strip('/')
+    if append_slash and remove_prefix:
+        remove_prefix += '/'
+    for file_path, fd in tar_generator:
+        assert file_path.startswith(remove_prefix)
+        output_path = os.path.join(output_dir, file_path[len(remove_prefix):].lstrip('/'))
+        os.makedirs(os.path.dirname(output_path), exist_ok=True)
+        with open(output_path, 'wb') as f:
+            f.write(fd.read())
+
+
 def download_file(path: str, url: str, update: bool = True):
     """Download a file over http[s]. With `update`, tries to use mtime timestamps to download only changed files."""
     url_time = None