Index

core

Core module for axm-edit.

Contains the batch editing engine and git checkpoint logic.

batch_apply(root, operations)

Validate and apply a batch of file operations.

Atomicity contract:

• Validation is the gate. All operations are validated first; if any fails the batch is rejected wholesale (success=False) before a single byte is written — a true all-or-nothing guarantee. Validation includes the syntax gate: a batch that would leave a parseable Python, TOML or JSON file unparseable is refused (:mod:axm_edit.core.syntax_gate). JSON comments and trailing commas are tolerated; new JSON files are unchecked.
• Apply is best-effort with automatic rollback. Once validation passes, a targeted checkpoint of every touched path is captured and the operations are applied. If any exception occurs mid-apply (anchor drift, a write_text / unlink / mkdir failure, a permission error, …) the partial work is rolled back to that checkpoint — touched files are restored to their pre-batch bytes and batch-created files (and the empty directories created for them) are removed — and a failing BatchResult is returned. The filesystem is not made transactional at the OS level; the rollback restores only the snapshotted paths.

Parameters:

Name	Type	Description	Default
root	Path	Project root directory (all paths are relative to this).	required
operations	Sequence[Operation]	List of replace, create, rewrite and delete operations.	required

Returns:

Type	Description
BatchResult	BatchResult with success status, a targeted-path snapshot
BatchResult	(checkpoint) for rollback, and a summary.

Source code in packages/axm-edit/src/axm_edit/core/engine.py

def batch_apply(root: Path, operations: Sequence[Operation]) -> BatchResult:
    """Validate and apply a batch of file operations.

    Atomicity contract:

    * **Validation is the gate.** All operations are validated first; if any
      fails the batch is rejected wholesale (``success=False``) before a
      single byte is written — a true all-or-nothing guarantee. Validation
      includes the syntax gate: a batch that would leave a parseable Python, TOML
      or JSON file unparseable is refused (:mod:`axm_edit.core.syntax_gate`).
      JSON comments and trailing commas are tolerated; new JSON files are
      unchecked.
    * **Apply is best-effort with automatic rollback.** Once validation
      passes, a targeted checkpoint of every touched path is captured and the
      operations are applied. If *any* exception occurs mid-apply (anchor
      drift, a ``write_text`` / ``unlink`` / ``mkdir`` failure, a permission
      error, …) the partial work is rolled back to that checkpoint — touched
      files are restored to their pre-batch bytes and batch-created files (and
      the empty directories created for them) are removed — and a failing
      ``BatchResult`` is returned. The filesystem is *not* made transactional
      at the OS level; the rollback restores only the snapshotted paths.

    Args:
        root: Project root directory (all paths are relative to this).
        operations: List of replace, create, rewrite and delete operations.

    Returns:
        BatchResult with success status, a targeted-path snapshot
        (``checkpoint``) for rollback, and a summary.
    """
    root = root.resolve()
    grouped = _group_operations(root, operations)
    resolved_by_file, errors = _validate_all(root, grouped)
    if not errors:
        errors = _syntax_errors(root, grouped, resolved_by_file)

    if errors:
        return BatchResult(
            success=False,
            error="Validation failed",
            details=errors,
        )

    checkpoint = create_checkpoint(root, operations)

    try:
        total_applied = 0
        for file_rel, resolved in resolved_by_file.items():
            total_applied += _apply_replace(root, file_rel, resolved)
        total_applied += _apply_creates_deletes(root, grouped.creates, grouped.deletes)
        total_applied += _apply_rewrites(root, grouped.rewrites)
    except _AnchorDriftError as drift:
        rb = rollback(root, checkpoint)
        return BatchResult(
            success=False,
            checkpoint=checkpoint,
            error="Apply aborted: file drifted between validation and apply",
            details=[drift.detail],
            rollback_failed=not rb.ok,
        )
    except Exception as exc:  # noqa: BLE001 - any apply failure must roll back
        rb = rollback(root, checkpoint)
        return BatchResult(
            success=False,
            checkpoint=checkpoint,
            error=f"Apply aborted and rolled back: {exc}",
            rollback_failed=not rb.ok,
        )

    summary = {
        "modified": len(resolved_by_file),
        "created": len(grouped.creates),
        "deleted": len(grouped.deletes),
    }
    if grouped.rewrites:
        summary["rewritten"] = len(grouped.rewrites)

    return BatchResult(
        success=True,
        checkpoint=checkpoint,
        applied=total_applied,
        summary=summary,
    )

create_checkpoint(root, operations)

Snapshot every path operations will touch, before they are applied.

For each operation's target path the snapshot records whether the file currently exists and, if so, its original bytes. The result is a JSON string keyed by resolved relative path, suitable for storage on BatchResult.checkpoint and for passing back to :func:rollback.

Parameters:

Name	Type	Description	Default
root	Path	Project root directory (all paths are relative to this).	required
operations	Sequence[Operation]	The batch about to be applied — replace, create and delete operations whose file attribute names the target.	required

Returns:

Type	Description
str	A JSON snapshot string. Always returned (never None) whenever
str	there are operations, in git and non-git directories alike.

Source code in packages/axm-edit/src/axm_edit/core/checkpoint.py

def create_checkpoint(root: Path, operations: Sequence[Operation]) -> str:
    """Snapshot every path *operations* will touch, before they are applied.

    For each operation's target path the snapshot records whether the file
    currently exists and, if so, its original bytes. The result is a JSON
    string keyed by resolved relative path, suitable for storage on
    ``BatchResult.checkpoint`` and for passing back to :func:`rollback`.

    Args:
        root: Project root directory (all paths are relative to this).
        operations: The batch about to be applied — replace, create and
            delete operations whose ``file`` attribute names the target.

    Returns:
        A JSON snapshot string. Always returned (never ``None``) whenever
        there are operations, in git and non-git directories alike.
    """
    root = root.resolve()
    entries: dict[str, str | None] = {}
    created_dirs: set[str] = set()
    for op in operations:
        target = _resolve_within(root, op.file)
        if target is None:
            continue
        # Key the dedup on the canonical resolved-within path, not the raw
        # spelling: "a.py" and "./a.py" name the same file and must collapse
        # to a single entry. The canonical key is also what rollback re-resolves.
        rel = target.relative_to(root).as_posix()
        if rel in entries:
            continue
        if target.is_file():
            entries[rel] = base64.b64encode(target.read_bytes()).decode("ascii")
        else:
            entries[rel] = None
            created_dirs |= _ancestors_to_create(root, target)
    return json.dumps(
        {
            "version": _SNAPSHOT_VERSION,
            "entries": entries,
            "created_dirs": sorted(created_dirs),
        }
    )

rollback(root, checkpoint)

Restore exactly the paths captured by checkpoint to their prior state.

Rollback is a strict inverse of the batch and best-effort: for each snapshotted path a file that existed is rewritten with its original bytes, a file that did not exist before is removed, and only the directories the batch itself created (recorded in the snapshot) are pruned — a pre-existing directory is never removed. Every captured path is attempted even if an earlier one fails, so a partial rollback is fully reported. No git command is run.

Parameters:

Name	Type	Description	Default
root	Path	Project root directory.	required
checkpoint	str	The JSON snapshot returned by :func:create_checkpoint.	required

Returns:

Name	Type	Description
A	RollbackResult	class:~axm_edit.models.operations.RollbackResult listing the
	RollbackResult	paths restored and those that could not be restored. RollbackResult.ok
	RollbackResult	is True only on a well-formed snapshot with no per-path failure;
	RollbackResult	a malformed snapshot yields valid=False.

Source code in packages/axm-edit/src/axm_edit/core/checkpoint.py

def rollback(root: Path, checkpoint: str) -> RollbackResult:
    """Restore exactly the paths captured by *checkpoint* to their prior state.

    Rollback is a *strict inverse* of the batch and best-effort: for each
    snapshotted path a file that existed is rewritten with its original bytes,
    a file that did not exist before is removed, and only the directories the
    batch itself created (recorded in the snapshot) are pruned — a
    pre-existing directory is never removed. Every captured path is attempted
    even if an earlier one fails, so a partial rollback is fully reported. No
    git command is run.

    Args:
        root: Project root directory.
        checkpoint: The JSON snapshot returned by :func:`create_checkpoint`.

    Returns:
        A :class:`~axm_edit.models.operations.RollbackResult` listing the
        paths restored and those that could not be restored. ``RollbackResult.ok``
        is ``True`` only on a well-formed snapshot with no per-path failure;
        a malformed snapshot yields ``valid=False``.
    """
    root = root.resolve()
    try:
        payload = json.loads(checkpoint)
        entries = payload["entries"]
    except (ValueError, TypeError, KeyError):
        return RollbackResult(valid=False)
    if not isinstance(entries, dict):
        return RollbackResult(valid=False)

    created_dirs = _read_created_dirs(payload)
    restored: list[str] = []
    unrestored: list[str] = []
    for rel, encoded in entries.items():
        target = _resolve_within(root, rel)
        if target is None:
            continue
        try:
            _restore_one(target, encoded, root, created_dirs)
            restored.append(rel)
        except OSError:
            unrestored.append(rel)
    return RollbackResult(restored=restored, unrestored=unrestored)