# Copyright 2025–2026 European Union
# Author: Bulgheroni Antonio (antonio.bulgheroni@ec.europa.eu)
# SPDX-License-Identifier: EUPL-1.2
"""
Atomic, roll-back-safe writer for generated projects.
The writer renders a project into a temporary *staging* directory created on the
same filesystem as the final destination, and only moves it into place once
rendering has completed successfully. If anything fails — a rendering error, a
permission problem, or an interruption — the staging directory is removed and any
destination directory the writer created is deleted, leaving the filesystem in its
exact pre-run state.
This guarantees two of the feature's correctness properties: no partially
generated project is ever left behind, and a non-empty destination is never
touched.
.. versionadded:: 2.4
"""
from __future__ import annotations
import shutil
import tempfile
from collections.abc import Callable
from pathlib import Path
from mafw.scaffold.errors import DestinationNotEmptyError, GenerationError
from mafw.scaffold.spec import ProjectSpec
[docs]
def _is_non_empty_dir(path: Path) -> bool:
"""Return ``True`` if ``path`` exists, is a directory, and contains any entry.
:param path: The path to inspect.
:type path: Path
:return: Whether the path is a non-empty directory.
:rtype: bool
"""
return path.is_dir() and any(path.iterdir())
[docs]
def write_atomically(spec: ProjectSpec, render: Callable[[Path], None]) -> Path:
"""Render a project into staging and atomically move it to the destination.
The ``render`` callable receives the staging directory and is expected to
populate it with the full project tree. The staging directory is created next
to the destination (same filesystem) so the final move is an atomic rename
rather than a cross-device copy.
:param spec: The validated project specification; ``spec.destination`` is the
final project directory.
:type spec: mafw.scaffold.spec.ProjectSpec
:param render: A callable that writes the project into the given staging dir.
:type render: Callable[[Path], None]
:return: The absolute path to the generated project root.
:rtype: Path
:raises mafw.scaffold.errors.DestinationNotEmptyError: If the destination already contains entries.
:raises PermissionError: If the destination cannot be written.
:raises mafw.scaffold.errors.GenerationError: If rendering fails; the filesystem is rolled back first.
"""
destination = spec.destination
# Overwrite protection is checked before any staging work so a non-empty
# destination is never modified.
if _is_non_empty_dir(destination):
raise DestinationNotEmptyError(f'The destination {destination} is not empty; refusing to overwrite it.')
# The staging directory must live on the same filesystem as the destination so
# the final move is atomic. We create it inside the destination's parent.
staging_parent = destination.parent
# Creating the parent may be needed; track whether we created the destination
# directory ourselves so rollback can remove exactly what we added.
created_parent = not staging_parent.exists()
staging_parent.mkdir(parents=True, exist_ok=True)
destination_existed = destination.exists()
staging_dir = Path(tempfile.mkdtemp(prefix='.mafw-init-', dir=staging_parent))
try:
# Populate the staging tree. Any failure here triggers full rollback.
render(staging_dir)
# Move the rendered tree into place. If the (empty) destination already
# exists, move the staging *contents* into it; otherwise rename the
# staging dir to the destination.
if destination_existed:
for entry in staging_dir.iterdir():
shutil.move(str(entry), str(destination / entry.name))
staging_dir.rmdir()
else:
staging_dir.replace(destination)
except DestinationNotEmptyError:
raise
except PermissionError:
_rollback(staging_dir, destination, destination_existed, created_parent, staging_parent)
raise
except BaseException as exc:
# BaseException also covers KeyboardInterrupt, so an interrupt mid-render
# still rolls back cleanly before the exception propagates.
_rollback(staging_dir, destination, destination_existed, created_parent, staging_parent)
if isinstance(exc, Exception):
raise GenerationError(f'Project generation failed: {exc}') from exc
raise
return destination
[docs]
def _rollback(
staging_dir: Path,
destination: Path,
destination_existed: bool,
created_parent: bool,
staging_parent: Path,
) -> None:
"""Undo any filesystem changes made during a failed generation.
:param staging_dir: The staging directory to remove.
:type staging_dir: Path
:param destination: The final destination directory.
:type destination: Path
:param destination_existed: Whether the destination existed before this run.
:type destination_existed: bool
:param created_parent: Whether this run created the destination's parent.
:type created_parent: bool
:param staging_parent: The destination's parent directory.
:type staging_parent: Path
"""
# Always remove the staging directory.
shutil.rmtree(staging_dir, ignore_errors=True)
# Remove the destination only if this run created it (it did not pre-exist).
if not destination_existed and destination.exists():
shutil.rmtree(destination, ignore_errors=True)
# Remove a parent directory we created, but only if it is now empty.
if created_parent and staging_parent.exists() and not any(staging_parent.iterdir()):
staging_parent.rmdir()