From f963845143fe19224c23f61ecae01cbf9c90d948 Mon Sep 17 00:00:00 2001 From: "Thomas G. Close" Date: Fri, 16 Oct 2026 15:44:10 +1100 Subject: [PATCH 1/3] add generic staged_out_dir wrapper --- fileformats/core/__init__.py | 27 +++++++++++------- fileformats/core/extras.py | 54 ++++++++++++++++++++++++++++++++++++ 2 files changed, 71 insertions(+), 10 deletions(-) diff --git a/fileformats/core/__init__.py b/fileformats/core/__init__.py index 1431ee0e..ab839e91 100644 --- a/fileformats/core/__init__.py +++ b/fileformats/core/__init__.py @@ -3,7 +3,13 @@ from .convertible import ConvertibleTo, ConvertibleToMarker from .datatype import DataType, FieldPrimitive from .decorators import mtime_cached_property, validated_property -from .extras import converter, extra, extra_implementation, find_extra_implementation +from .extras import ( + converter, + extra, + extra_implementation, + find_extra_implementation, + staged_out_dir, +) from .field import Field from .fileset import FileSet, FileSetMetadata, FileSetPrimitive from .identification import find_matching, from_mime, from_paths, to_mime @@ -12,9 +18,7 @@ from .sampling import SampleFileGenerator __all__ = [ - "__version__", "Classifier", - "converter", "ConvertibleTo", "ConvertibleToMarker", "DataType", @@ -23,18 +27,21 @@ "FileSet", "FileSetMetadata", "FileSetPrimitive", - "find_extra_implementation", - "find_matching", - "from_mime", - "from_paths", "Loaded", "LoadedMarker", "MockMixin", - "mtime_cached_property", "SampleFileGenerator", - "to_mime", - "validated_property", + "__version__", "check_loaded", + "converter", "extra", "extra_implementation", + "find_extra_implementation", + "find_matching", + "from_mime", + "from_paths", + "mtime_cached_property", + "staged_out_dir", + "to_mime", + "validated_property", ] diff --git a/fileformats/core/extras.py b/fileformats/core/extras.py index e0bb3bbc..47da4806 100644 --- a/fileformats/core/extras.py +++ b/fileformats/core/extras.py @@ -2,10 +2,14 @@ import importlib import inspect import logging +import os +import shutil +import tempfile import types import typing as ty import urllib.error from itertools import zip_longest +from pathlib import Path from typing import Annotated, Self import fileformats.core @@ -116,6 +120,56 @@ def decorated_extra(obj: DataType, *args: ty.Any, **kwargs: ty.Any) -> ty.Any: return decorated_extra # type: ignore[return-value] +def staged_out_dir( + implementation: ty.Callable[..., ty.Any], + obj: ty.Any, + out_dir: "os.PathLike[str] | str", + *args: ty.Any, + **kwargs: ty.Any, +) -> ty.Any: + """A wrapper for extra methods that write a fileset to the `out_dir` passed as + their first argument (e.g. ``@extra(wrapper=staged_out_dir)``), which passes the + implementations a hidden staging directory alongside `out_dir` instead, and only + moves its contents into `out_dir` once they have returned. That way, only complete + output ends up in `out_dir`, and implementations are free to copy files there and + modify them in place. + + The implementations need to return a fileset within the `out_dir` they are passed, + which is returned relocated to the requested `out_dir`. Any existing files in + `out_dir` are left in place, but a FileExistsError is raised, before anything is + moved, if the output would overwrite any of them. + """ + out_dir = Path(out_dir).absolute() + out_dir.parent.mkdir(parents=True, exist_ok=True) + staging_dir = Path( + tempfile.mkdtemp(dir=out_dir.parent, prefix=f".staging-{out_dir.name}-") + ) + try: + output = implementation(obj, staging_dir, *args, **kwargs) + try: + rel_fspaths = [ + p.absolute().relative_to(staging_dir) for p in output.fspaths + ] + except ValueError: + raise RuntimeError( + f"The {getattr(implementation, '__name__', implementation)!r} " + f"implementation for {type(obj)} didn't write its output to the " + f"'out_dir' it was passed: {output.fspaths}" + ) from None + entries = sorted(staging_dir.iterdir()) + if existing := [e.name for e in entries if (out_dir / e.name).exists()]: + raise FileExistsError( + f"Cannot write the output for {obj} into {out_dir} as {existing} " + "already exist(s) there" + ) + out_dir.mkdir(exist_ok=True) + for entry in entries: + os.replace(entry, out_dir / entry.name) + finally: + shutil.rmtree(staging_dir, ignore_errors=True) + return type(output)([out_dir / p for p in rel_fspaths]) + + def _import_base_extras_modules(datatype: ty.Type[DataType]) -> None: """Import the extras modules of the base classes of `datatype`, which may contain implementations it inherits (e.g. if it subclasses a format from another namespace From 2ba7b925237d8dd4aebe2f995fa0be23c365042b Mon Sep 17 00:00:00 2001 From: "Thomas G. Close" Date: Fri, 16 Oct 2026 15:45:18 +1100 Subject: [PATCH 2/3] Adds FileSet.map_nested method for running a method on all nested filesets (e.g. deidentify) --- fileformats/core/fileset.py | 101 +++++++++++- fileformats/core/tests/test_map_nested.py | 187 ++++++++++++++++++++++ 2 files changed, 281 insertions(+), 7 deletions(-) create mode 100644 fileformats/core/tests/test_map_nested.py diff --git a/fileformats/core/fileset.py b/fileformats/core/fileset.py index f2810283..fe9d8226 100644 --- a/fileformats/core/fileset.py +++ b/fileformats/core/fileset.py @@ -578,22 +578,103 @@ def required_paths(self) -> ty.FrozenSet[Path]: required.add(path) return frozenset(required) - def nested_filesets(self) -> ty.List["FileSet"]: - """Returns all nested filesets that are required for the format + def nested_filesets(self, recursive: bool = True) -> ty.List["FileSet"]: + """Returns all nested filesets that are required for the format, i.e. those + returned by its validated properties, including those within dicts and + sequences returned by them + + Parameters + ---------- + recursive : bool, optional + whether to also include the filesets nested within the nested filesets, by + default True Returns ------ fileset : list[FileSet] a fileset that is nested within the broader fileset """ - nested = [] + nested: ty.List[FileSet] = [] + + def collect(value: ty.Any) -> None: + if isinstance(value, FileSet): + nested.append(value) + if recursive: + nested.extend(value.nested_filesets()) + elif isinstance(value, ty.Mapping): + for v in value.values(): + collect(v) + elif isinstance(value, (list, tuple, set, frozenset)): + for v in value: + collect(v) + for prop_name in sorted(self.validated_properties()): - prop = getattr(self, prop_name) - if isinstance(prop, FileSet): - nested.append(prop) - nested.extend(prop.nested_filesets()) + collect(getattr(self, prop_name)) return nested + def map_nested( + self, + out_dir: "os.PathLike[str] | str", + func: ty.Callable[["FileSet", Path], ty.Optional["FileSet"]], + ) -> Self: + """Writes a copy of the fileset to `out_dir`, in which the filesets nested + directly within it (see `nested_filesets`) are written by `func` and its + remaining files are copied unchanged, keeping the same layout relative to their + common parent directory. For example, to write converted or otherwise modified + versions of the nested filesets. + + Nested filesets that point outside of the fileset (i.e. aren't made up of its + paths) aren't part of it, so are left out of the copy, as they are by `copy` + + Parameters + ---------- + out_dir : PathLike[str] | str + the directory to write the copy to + func : Callable[[FileSet, Path], FileSet | None] + called with each nested fileset and the directory it should be written to, + to be in the same position relative to the copy as the nested fileset is + to this fileset. It should write its counterpart there with the same + layout (e.g. names, so that the copy is still of this fileset's type) and + return it, or return None to have the nested fileset copied unchanged along + with the rest of the files. Nested filesets that are entirely within ones + already written by `func` aren't passed to it. + + Returns + ------- + Self + the copy of the fileset + + Raises + ------ + ValueError + if a nested fileset is made up of paths both within and outside the fileset + """ + out_dir = Path(out_dir) + src_root = _root_dir(self) + own_paths = set(self.all_file_paths()) + written: ty.Set[Path] = set() + for nested in self.nested_filesets(recursive=False): + nested_paths = set(nested.all_file_paths()) + if outside := sorted(nested_paths - own_paths): + if outside == sorted(nested_paths): + continue # not part of the fileset + raise ValueError( + f"{nested} is nested within {self}, but is made up of paths both " + f"within it and outside it, {outside}" + ) + if nested_paths <= written: + continue + dest_dir = out_dir / _root_dir(nested).relative_to(src_root) + if func(nested, dest_dir) is not None: + written.update(nested_paths) + for path in own_paths - written: + dest = out_dir / path.relative_to(src_root) + dest.parent.mkdir(parents=True, exist_ok=True) + shutil.copy2(path, dest) + # Reinstantiating the type checks that `func` preserved the layout it requires + # (e.g. didn't rename files it looks for) + return type(self)([out_dir / p.relative_to(src_root) for p in self.fspaths]) + def trim_paths(self) -> Self: """Trims paths in fspaths to only those that are "required" by the format class i.e. returned by a required property""" @@ -2309,3 +2390,9 @@ def _destination_to_avoid( _formats_by_name: ty.Optional[ty.Dict[str, ty.Set[ty.Type["FileSet"]]]] = None _required_props: ty.Optional[ty.Tuple[str, ...]] = None _valid_class: ty.Optional[bool] = None + + +def _root_dir(fileset: FileSet) -> Path: + """The directory containing all of the fileset's paths (the parent of a directory + fileset rather than the directory itself)""" + return Path(os.path.commonpath([p.parent for p in fileset.fspaths])) diff --git a/fileformats/core/tests/test_map_nested.py b/fileformats/core/tests/test_map_nested.py new file mode 100644 index 00000000..e698df38 --- /dev/null +++ b/fileformats/core/tests/test_map_nested.py @@ -0,0 +1,187 @@ +import os +import typing as ty +from pathlib import Path + +import pytest + +from fileformats.core import ( + FileSet, + extra, + extra_implementation, + staged_out_dir, + validated_property, +) +from fileformats.core.exceptions import FormatMismatchError +from fileformats.generic import Directory, File + + +class Leaf(File): + ext = ".leaf" + + +class Inner(Directory): + @validated_property + def leaves(self) -> ty.List[Leaf]: + found = [Leaf(p) for p in sorted(self.fspath.glob("*.leaf"))] + if not found: + raise FormatMismatchError(f"No leaves in {self.fspath}") + return found + + +class Outer(Directory): + @validated_property + def inners(self) -> ty.Dict[str, Inner]: + return {p.name: Inner(p) for p in sorted(self.fspath.glob("inner*"))} + + @validated_property + def top_leaf(self) -> Leaf: + return Leaf(self.fspath / "top.leaf") + + +class WithExternal(Directory): + @validated_property + def external(self) -> Leaf: + return Leaf(self.fspath.parent / "external.leaf") + + +def _make_outer(root: Path) -> Outer: + outer = root / "outer" + for inner in ("inner1", "inner2"): + (outer / inner).mkdir(parents=True) + (outer / inner / "a.leaf").write_text("original") + (outer / inner / "notes.txt").write_text("notes") + (outer / "top.leaf").write_text("original") + (outer / ".hidden").write_text("hidden") + return Outer(outer) + + +def _contents(fileset: FileSet, root: Path) -> ty.Dict[str, str]: + return {str(p.relative_to(root)): p.read_text() for p in fileset.all_file_paths()} + + +def _modify(nested: FileSet, dest_dir: Path) -> ty.Optional[FileSet]: + """Writes modified copies of Leaf files, and leaves the rest to be copied""" + if isinstance(nested, Leaf): + dest_dir.mkdir(parents=True, exist_ok=True) + dest = dest_dir / nested.fspath.name + dest.write_text("modified") + return Leaf(dest) + if isinstance(nested, Inner): + return nested.map_nested(dest_dir, _modify) + return None + + +def test_nested_filesets_within_dicts_and_lists(tmp_path: Path) -> None: + outer = _make_outer(tmp_path) + direct = outer.nested_filesets(recursive=False) + assert [type(n).__name__ for n in direct] == ["Inner", "Inner", "Leaf"] + assert [type(n).__name__ for n in outer.nested_filesets()] == [ + "Inner", + "Leaf", + "Inner", + "Leaf", + "Leaf", + ] + + +def test_map_nested(tmp_path: Path) -> None: + out_dir = tmp_path / "out" + mapped = _make_outer(tmp_path / "in").map_nested(out_dir, _modify) + + assert isinstance(mapped, Outer) + assert mapped.fspath == out_dir / "outer" + assert _contents(mapped, out_dir) == { + "outer/.hidden": "hidden", + "outer/inner1/a.leaf": "modified", + "outer/inner1/notes.txt": "notes", + "outer/inner2/a.leaf": "modified", + "outer/inner2/notes.txt": "notes", + "outer/top.leaf": "modified", + } + + +def test_map_nested_none_copies_unchanged(tmp_path: Path) -> None: + outer = _make_outer(tmp_path / "in") + mapped = outer.map_nested(tmp_path / "out", lambda nested, dest_dir: None) + assert _contents(mapped, tmp_path / "out") == _contents(outer, tmp_path / "in") + + +def test_map_nested_skips_external(tmp_path: Path) -> None: + (tmp_path / "in" / "dir").mkdir(parents=True) + (tmp_path / "in" / "dir" / "own.txt").write_text("own") + (tmp_path / "in" / "external.leaf").write_text("external") + called = [] + + def func(nested: FileSet, dest_dir: Path) -> None: + called.append(nested) + + with pytest.raises(FileNotFoundError, match="external.leaf"): + # the copy doesn't include the external file, so isn't a valid WithExternal + WithExternal(tmp_path / "in" / "dir").map_nested(tmp_path / "out", func) + assert called == [] + assert sorted(p.name for p in (tmp_path / "out").iterdir()) == ["dir"] + assert not (tmp_path / "out" / "external.leaf").exists() + + +def test_map_nested_checks_layout(tmp_path: Path) -> None: + def rename(nested: FileSet, dest_dir: Path) -> ty.Optional[FileSet]: + if isinstance(nested, Leaf) and nested.fspath.name == "top.leaf": + dest_dir.mkdir(parents=True, exist_ok=True) + (dest_dir / "renamed.leaf").write_text("modified") + return Leaf(dest_dir / "renamed.leaf") + return None + + with pytest.raises(FileNotFoundError, match="top.leaf"): + _make_outer(tmp_path / "in").map_nested(tmp_path / "out", rename) + + +class Staged(File): + ext = ".staged" + + @extra(wrapper=staged_out_dir) + def write_copy(self, out_dir: "os.PathLike[str]", fail: bool = False) -> "Staged": + raise NotImplementedError + + +@extra_implementation(Staged.write_copy) +def staged_write_copy( + staged: Staged, out_dir: "os.PathLike[str]", fail: bool = False +) -> Staged: + out_path = Path(out_dir) / staged.fspath.name + out_path.write_text("written") + if fail: + raise RuntimeError("failed") + return Staged(out_path) + + +def test_staged_out_dir(tmp_path: Path) -> None: + (tmp_path / "in").mkdir() + (tmp_path / "in" / "a.staged").write_text("original") + out_dir = tmp_path / "out" + written = Staged(tmp_path / "in" / "a.staged").write_copy(out_dir) + + assert written.fspath == out_dir / "a.staged" + assert written.fspath.read_text() == "written" + # the staging directory is cleaned up + assert sorted(p.name for p in tmp_path.iterdir()) == ["in", "out"] + + +def test_staged_out_dir_failure_leaves_nothing(tmp_path: Path) -> None: + (tmp_path / "in").mkdir() + (tmp_path / "in" / "a.staged").write_text("original") + out_dir = tmp_path / "out" + with pytest.raises(RuntimeError, match="failed"): + Staged(tmp_path / "in" / "a.staged").write_copy(out_dir, fail=True) + assert not out_dir.exists() + assert sorted(p.name for p in tmp_path.iterdir()) == ["in"] + + +def test_staged_out_dir_wont_overwrite(tmp_path: Path) -> None: + (tmp_path / "in").mkdir() + (tmp_path / "in" / "a.staged").write_text("original") + out_dir = tmp_path / "out" + out_dir.mkdir() + (out_dir / "a.staged").write_text("existing") + with pytest.raises(FileExistsError): + Staged(tmp_path / "in" / "a.staged").write_copy(out_dir) + assert (out_dir / "a.staged").read_text() == "existing" From c32bfd5639b86eaa51703013a41e4622fa042698 Mon Sep 17 00:00:00 2001 From: "Thomas G. Close" Date: Fri, 16 Oct 2026 16:34:04 +1100 Subject: [PATCH 3/3] add windows support to map nested tests --- fileformats/core/tests/test_map_nested.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/fileformats/core/tests/test_map_nested.py b/fileformats/core/tests/test_map_nested.py index e698df38..b7755a31 100644 --- a/fileformats/core/tests/test_map_nested.py +++ b/fileformats/core/tests/test_map_nested.py @@ -56,7 +56,10 @@ def _make_outer(root: Path) -> Outer: def _contents(fileset: FileSet, root: Path) -> ty.Dict[str, str]: - return {str(p.relative_to(root)): p.read_text() for p in fileset.all_file_paths()} + # Keys use "/" separators on all platforms (incl. Windows) + return { + p.relative_to(root).as_posix(): p.read_text() for p in fileset.all_file_paths() + } def _modify(nested: FileSet, dest_dir: Path) -> ty.Optional[FileSet]: