Skip to content

Filesystem

filesystem

FileSystem Adapter — atomic filesystem operations with rollback.

Provides safe file/directory creation with transaction support for atomic multi-file operations used during project scaffolding.

FileSystemAdapter

Adapter for atomic filesystem operations.

Supports both single operations and transactions for multi-file atomic operations with rollback on failure.

Source code in packages/axm-init/src/axm_init/adapters/filesystem.py 
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
class FileSystemAdapter:
    """Adapter for atomic filesystem operations.

    Supports both single operations and transactions for multi-file
    atomic operations with rollback on failure.
    """

    def write_file(self, path: Path, content: str) -> bool:
        """Write content to a file, creating parent directories.

        Args:
            path: Target file path.
            content: File content to write.

        Returns:
            True if successful.
        """
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content)
        return True

    def create_dir(self, path: Path) -> bool:
        """Create a directory (and parents).

        Args:
            path: Directory path to create.

        Returns:
            True if successful.
        """
        path.mkdir(parents=True, exist_ok=True)
        return True

    @contextmanager
    def transaction(self) -> Generator[Transaction, None, None]:
        """Context manager for atomic multi-file operations.

        On success: files are kept.
        On exception: all created files are rolled back.

        Yields:
            Transaction object for tracked operations.
        """
        tx = Transaction()
        try:
            yield tx
            tx.commit()
        except Exception:
            tx.rollback()
            raise
create_dir(path)

Create a directory (and parents).

Parameters:

Name Type Description Default
path Path

Directory path to create.

 required

Returns:

Type Description
bool

True if successful.
Source code in packages/axm-init/src/axm_init/adapters/filesystem.py 
89
90
91
92
93
94
95
96
97
98
99
def create_dir(self, path: Path) -> bool:
    """Create a directory (and parents).

    Args:
        path: Directory path to create.

    Returns:
        True if successful.
    """
    path.mkdir(parents=True, exist_ok=True)
    return True
transaction()

Context manager for atomic multi-file operations.

On success: files are kept. On exception: all created files are rolled back.

Yields:

Type Description
Transaction

Transaction object for tracked operations.
Source code in packages/axm-init/src/axm_init/adapters/filesystem.py 
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
@contextmanager
def transaction(self) -> Generator[Transaction, None, None]:
    """Context manager for atomic multi-file operations.

    On success: files are kept.
    On exception: all created files are rolled back.

    Yields:
        Transaction object for tracked operations.
    """
    tx = Transaction()
    try:
        yield tx
        tx.commit()
    except Exception:
        tx.rollback()
        raise
write_file(path, content)

Write content to a file, creating parent directories.

Parameters:

Name Type Description Default
path Path

Target file path.

 required
content str

File content to write.

 required

Returns:

Type Description
bool

True if successful.
Source code in packages/axm-init/src/axm_init/adapters/filesystem.py 
75
76
77
78
79
80
81
82
83
84
85
86
87
def write_file(self, path: Path, content: str) -> bool:
    """Write content to a file, creating parent directories.

    Args:
        path: Target file path.
        content: File content to write.

    Returns:
        True if successful.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content)
    return True

Transaction dataclass

Transaction context for atomic filesystem operations.

Tracks created files/directories for rollback on failure.

Source code in packages/axm-init/src/axm_init/adapters/filesystem.py 
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
@dataclass
class Transaction:
    """Transaction context for atomic filesystem operations.

    Tracks created files/directories for rollback on failure.
    """

    created_files: list[Path] = field(default_factory=list)
    created_dirs: list[Path] = field(default_factory=list)
    _committed: bool = False

    def write_file(self, path: Path, content: str) -> bool:
        """Write file and track for potential rollback."""
        path.parent.mkdir(parents=True, exist_ok=True)
        path.write_text(content)
        self.created_files.append(path)
        return True

    def create_dir(self, path: Path) -> bool:
        """Create directory and track for potential rollback."""
        path.mkdir(parents=True, exist_ok=True)
        self.created_dirs.append(path)
        return True

    def commit(self) -> None:
        """Mark transaction as committed (no rollback)."""
        self._committed = True

    def rollback(self) -> None:
        """Remove all created files and directories."""
        if self._committed:
            return

        # Remove files first
        for f in reversed(self.created_files):
            try:
                if f.exists():
                    f.unlink()
            except OSError as exc:
                logger.warning("Rollback: failed to remove file %s: %s", f, exc)

        # Remove directories (reverse order for nested)
        for d in reversed(self.created_dirs):
            try:
                if d.exists() and not any(d.iterdir()):
                    d.rmdir()
            except OSError as exc:
                logger.warning("Rollback: failed to remove dir %s: %s", d, exc)
commit()

Mark transaction as committed (no rollback).

Source code in packages/axm-init/src/axm_init/adapters/filesystem.py 
42
43
44
def commit(self) -> None:
    """Mark transaction as committed (no rollback)."""
    self._committed = True
create_dir(path)

Create directory and track for potential rollback.

Source code in packages/axm-init/src/axm_init/adapters/filesystem.py 
36
37
38
39
40
def create_dir(self, path: Path) -> bool:
    """Create directory and track for potential rollback."""
    path.mkdir(parents=True, exist_ok=True)
    self.created_dirs.append(path)
    return True
rollback()

Remove all created files and directories.

Source code in packages/axm-init/src/axm_init/adapters/filesystem.py 
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
def rollback(self) -> None:
    """Remove all created files and directories."""
    if self._committed:
        return

    # Remove files first
    for f in reversed(self.created_files):
        try:
            if f.exists():
                f.unlink()
        except OSError as exc:
            logger.warning("Rollback: failed to remove file %s: %s", f, exc)

    # Remove directories (reverse order for nested)
    for d in reversed(self.created_dirs):
        try:
            if d.exists() and not any(d.iterdir()):
                d.rmdir()
        except OSError as exc:
            logger.warning("Rollback: failed to remove dir %s: %s", d, exc)
write_file(path, content)

Write file and track for potential rollback.

Source code in packages/axm-init/src/axm_init/adapters/filesystem.py 
29
30
31
32
33
34
def write_file(self, path: Path, content: str) -> bool:
    """Write file and track for potential rollback."""
    path.parent.mkdir(parents=True, exist_ok=True)
    path.write_text(content)
    self.created_files.append(path)
    return True