smith.infrastructure.filesystem

 1"""Atomic file-system adapter — write files transactionally or remove them."""
 2
 3import shutil
 4import tempfile
 5from pathlib import Path
 6
 7from smith.domain.value_objects import FileSpec
 8
 9
10class FileSystemError(Exception):
11    """Raised when an atomic file operation fails."""
12
13
14class AtomicFileSystem:
15    """File-system adapter that writes files atomically via a staging directory."""
16
17    def __init__(self, project_dir: Path) -> None:
18        """Initialise with the project root directory."""
19        self._project_dir = project_dir
20
21    def check_conflicts(self, paths: list[Path]) -> list[Path]:
22        """Return the subset of *paths* that already exist on disk."""
23        return [p for p in paths if (self._project_dir / p).exists()]
24
25    def write_atomic(self, specs: list[FileSpec]) -> None:
26        """Write all specs atomically; roll back on failure."""
27        if not specs:
28            return
29        staging = Path(tempfile.mkdtemp(dir=self._project_dir))
30        try:
31            for spec in specs:
32                dest = staging / spec.relative_path
33                dest.parent.mkdir(parents=True, exist_ok=True)
34                dest.write_bytes(spec.content)
35            for spec in specs:
36                final = self._project_dir / spec.relative_path
37                final.parent.mkdir(parents=True, exist_ok=True)
38                Path(staging / spec.relative_path).replace(final)
39        except Exception as err:
40            shutil.rmtree(staging, ignore_errors=True)
41            raise FileSystemError("Atomic write failed") from err
42        else:
43            shutil.rmtree(staging, ignore_errors=True)
44
45    def remove(self, paths: list[Path]) -> None:
46        """Remove the given paths from the project directory."""
47        for p in paths:
48            full = self._project_dir / p
49            if full.exists():
50                full.unlink()
51
52    def exists(self, paths: list[Path]) -> dict[Path, bool]:
53        """Return a mapping of each path to whether it exists on disk."""
54        return {p: (self._project_dir / p).exists() for p in paths}