Move

move

Atomic move pipeline: relocate top-level symbols between modules.

ImportCycleError

Bases: Exception

Raised when a move would introduce a new import cycle.

Source code in packages/axm-anvil/src/axm_anvil/core/plan.py

class ImportCycleError(Exception):
    """Raised when a move would introduce a new import cycle."""

    def __init__(self, cycle: list[str]) -> None:
        self.cycle = list(cycle)
        chain = " \u2192 ".join([*self.cycle, self.cycle[0]])
        super().__init__(f"Import cycle: {chain}")

MovePathError

Bases: Exception

Raised when source and target paths share no usable common base.

Computing the relative paths handed to batch_edit requires a base directory that contains both the source and the target. When the two live in disjoint trees (e.g. different drives) no such base exists, so the fallback raises this typed error instead of leaking a bare ValueError.

Source code in packages/axm-anvil/src/axm_anvil/core/plan.py

class MovePathError(Exception):
    """Raised when source and target paths share no usable common base.

    Computing the relative paths handed to ``batch_edit`` requires a base
    directory that contains *both* the source and the target. When the two
    live in disjoint trees (e.g. different drives) no such base exists, so the
    fallback raises this typed error instead of leaking a bare ``ValueError``.
    """

    def __init__(self, source: object, target: object) -> None:
        self.source = source
        self.target = target
        super().__init__(
            f"No common base directory contains both {source} and {target}"
        )

MoveValidationError

Bases: Exception

Raised when a rendered module fails to parse post-transform.

Source code in packages/axm-anvil/src/axm_anvil/core/plan.py

class MoveValidationError(Exception):
    """Raised when a rendered module fails to parse post-transform."""

    def __init__(self, text: str, cause: BaseException) -> None:
        super().__init__(f"Rendered module failed to parse: {cause}")
        self.text = text

OverloadPartialMoveError

Bases: Exception

Raised when only a subset of an overload group is requested.

Source code in packages/axm-anvil/src/axm_anvil/core/plan.py

class OverloadPartialMoveError(Exception):
    """Raised when only a subset of an overload group is requested."""

SymbolAlreadyExistsError

Bases: Exception

Raised when a requested symbol already exists in the target module.

Source code in packages/axm-anvil/src/axm_anvil/core/plan.py

class SymbolAlreadyExistsError(Exception):
    """Raised when a requested symbol already exists in the target module."""

SymbolNotFoundError

Bases: Exception

Raised when a requested symbol does not exist in the source module.

Source code in packages/axm-anvil/src/axm_anvil/core/plan.py

class SymbolNotFoundError(Exception):
    """Raised when a requested symbol does not exist in the source module."""

batch_edit(path, operations)

Apply a batch of file operations atomically via axm-edit.

Accepts dict-shaped operations ({op, file, edits|content}) and delegates to :func:axm_edit.core.engine.batch_apply. Raises on any validation error so callers can trigger rollback.

Source code in packages/axm-anvil/src/axm_anvil/core/move.py

def batch_edit(  # type: ignore[explicit-any]  # JSON-shape payload at axm-edit frontier
    path: str | Path, operations: list[dict[str, Any]]
) -> None:
    """Apply a batch of file operations atomically via ``axm-edit``.

    Accepts dict-shaped operations (``{op, file, edits|content}``) and
    delegates to :func:`axm_edit.core.engine.batch_apply`. Raises on any
    validation error so callers can trigger rollback.
    """
    from axm_edit.core.engine import batch_apply
    from axm_edit.models.operations import CreateOp, DeleteOp, Edit, ReplaceOp

    ops: list[CreateOp | DeleteOp | ReplaceOp] = []
    for op in operations:
        kind = op.get("op")
        if kind == "replace":
            edits = op["edits"]
            if len(edits) == 1 and edits[0].get("old", "") == "":
                ops.append(
                    CreateOp(
                        file=op["file"],
                        content=edits[0]["new"],
                        overwrite=True,
                    )
                )
            else:
                ops.append(
                    ReplaceOp(
                        file=op["file"],
                        edits=[Edit(**e) for e in edits],
                    )
                )
        elif kind == "create":
            ops.append(
                CreateOp(
                    file=op["file"],
                    content=op["content"],
                    overwrite=bool(op.get("overwrite", False)),
                )
            )
        elif kind == "delete":
            ops.append(DeleteOp(file=op["file"]))
        else:
            raise ValueError(f"Unknown op kind: {kind!r}")
    result = batch_apply(Path(path), ops)
    if not result.success:
        raise RuntimeError(f"batch_edit failed: {result.error}")

detect_fixture_dependencies(blocks, local_names)

Return the pytest fixture names a set of moved blocks depend on.

A fixture dependency is a parameter name on a def test_* function or a @pytest.fixture-decorated function (including methods of a moved class), excluding self/cls, defaulted parameters, the pytest builtin fixtures in :data:PYTEST_BUILTIN_FIXTURES, and any name already resolvable as a local definition or import (local_names). Pure, in-memory CST analysis; no filesystem access.

Source code in packages/axm-anvil/src/axm_anvil/core/move.py

def detect_fixture_dependencies(blocks: list[Block], local_names: set[str]) -> set[str]:
    """Return the pytest fixture names a set of moved ``blocks`` depend on.

    A fixture dependency is a parameter name on a ``def test_*`` function or a
    ``@pytest.fixture``-decorated function (including methods of a moved
    class), excluding ``self``/``cls``, defaulted parameters, the pytest
    builtin fixtures in :data:`PYTEST_BUILTIN_FIXTURES`, and any name already
    resolvable as a local definition or import (``local_names``). Pure,
    in-memory CST analysis; no filesystem access.
    """
    used: set[str] = set()
    for block in blocks:
        for func in _function_defs_in_block(block):
            if not _is_fixture_consuming_function(func):
                continue
            for name in _candidate_fixture_params(func):
                if name in PYTEST_BUILTIN_FIXTURES or name in local_names:
                    continue
                used.add(name)
    return used

move_symbols(source_path, target_path, symbol_names, *, dry_run=False, workspace_root=None, shared_helpers='duplicate', shared_helpers_module=None, reexport=False, rename=None, check=False, strict=False, insert_after=None, include_helpers=True, side_effect_decorators=None)

Move top-level symbols from source_path to target_path.

Pipeline: parse → expand overloads → extract blocks → gather deps → build new target (imports + constants + symbols) → remove from source → classify shared helpers → validate parseability → atomic write via batch_edit → ruff fix.

shared_helpers selects the strategy when a helper is used by both a moved symbol and a remaining source symbol: "duplicate" copies and keeps the helper (emitting a warning); "error" aborts with :class:SharedHelpersError; "extract" is reserved for Phase 3.

When reexport=True, callers are left untouched and a from new_module import <names> # re-export for backwards compat line is appended to the source module. Incompatible with rename=.

When check=True, the move is simulated (no files written) and any newly introduced import cycle raises :class:ImportCycleError. A normal (non-dry_run) write also performs this check; dry_run=True alone preserves its historical "preview without enforcement" contract.

A requested name that is absent from the source module's top-level symbols is skipped with a warning on :attr:MovePlan.warnings rather than aborting the whole plan. Pass strict=True to restore the legacy behaviour of raising :class:SymbolNotFoundError on the first absent name.

insert_after controls where the moved blocks land in the target module body: when it names an existing top-level symbol the blocks are spliced immediately after it; when None (default) the blocks append at the end (unchanged contract); when it names an absent symbol the blocks append at the end and a warning is added to :attr:MovePlan.warnings. Imports and constants keep their historical placement regardless of insert_after.

include_helpers (default True) preserves the historical behaviour of copying transitively-referenced local helpers and constants into the target. When False those helpers/constants are not copied (the moved code is left referencing them), a warning enumerating the un-copied local helper names is added to :attr:MovePlan.warnings, and the shared_helpers classification is short-circuited (nothing is duplicated or extracted). Imports required by the moved code are always copied regardless of this flag.

Source code in packages/axm-anvil/src/axm_anvil/core/move.py

def move_symbols(  # noqa: PLR0913
    source_path: str | Path,
    target_path: str | Path,
    symbol_names: Sequence[str],
    *,
    dry_run: bool = False,
    workspace_root: Path | None = None,
    shared_helpers: str = "duplicate",
    shared_helpers_module: str | None = None,
    reexport: bool = False,
    rename: dict[str, str] | None = None,
    check: bool = False,
    strict: bool = False,
    insert_after: str | None = None,
    include_helpers: bool = True,
    side_effect_decorators: frozenset[str] | None = None,
) -> MovePlan:
    """Move top-level symbols from ``source_path`` to ``target_path``.

    Pipeline: parse → expand overloads → extract blocks → gather deps →
    build new target (imports + constants + symbols) → remove from source
    → classify shared helpers → validate parseability → atomic write via
    ``batch_edit`` → ruff fix.

    ``shared_helpers`` selects the strategy when a helper is used by both a
    moved symbol and a remaining source symbol: ``"duplicate"`` copies and
    keeps the helper (emitting a warning); ``"error"`` aborts with
    :class:`SharedHelpersError`; ``"extract"`` is reserved for Phase 3.

    When ``reexport=True``, callers are left untouched and a
    ``from new_module import <names>  # re-export for backwards compat`` line
    is appended to the source module. Incompatible with ``rename=``.

    When ``check=True``, the move is simulated (no files written) and any
    *newly introduced* import cycle raises :class:`ImportCycleError`. A
    normal (non-``dry_run``) write also performs this check; ``dry_run=True``
    alone preserves its historical "preview without enforcement" contract.

    A requested name that is absent from the source module's top-level
    symbols is **skipped** with a warning on :attr:`MovePlan.warnings`
    rather than aborting the whole plan. Pass ``strict=True`` to restore
    the legacy behaviour of raising :class:`SymbolNotFoundError` on the
    first absent name.

    ``insert_after`` controls where the moved *blocks* land in the target
    module body: when it names an existing top-level symbol the blocks are
    spliced immediately after it; when ``None`` (default) the blocks append
    at the end (unchanged contract); when it names an absent symbol the
    blocks append at the end and a warning is added to
    :attr:`MovePlan.warnings`. Imports and constants keep their historical
    placement regardless of ``insert_after``.

    ``include_helpers`` (default ``True``) preserves the historical
    behaviour of copying transitively-referenced local helpers and
    constants into the target. When ``False`` those helpers/constants are
    **not** copied (the moved code is left referencing them), a warning
    enumerating the un-copied local helper names is added to
    :attr:`MovePlan.warnings`, and the ``shared_helpers`` classification is
    short-circuited (nothing is duplicated or extracted). Imports required
    by the moved code are always copied regardless of this flag.
    """
    _validate_options(shared_helpers, shared_helpers_module, reexport, rename)

    source_path = Path(source_path)
    target_path = Path(target_path)

    source_text = source_path.read_text()
    target_text = target_path.read_text()
    source_tree = cst.parse_module(source_text)
    target_tree = cst.parse_module(target_text)

    expanded_names, moved_names, skipped_warnings = _validate_and_expand(
        source_tree, target_tree, symbol_names, strict=strict
    )
    if not moved_names:
        # Nothing resolved to a present symbol (all absent, non-strict): the
        # move is a no-op. Return an empty plan with the original texts BEFORE
        # any tree-building or write, so source and target stay byte-identical.
        noop_plan = _build_plan(
            MoveContext(
                source_text_new=source_text,
                target_text_new=target_text,
                moved_names=moved_names,
                imports_added=[],
                constants_added=[],
                shared_map={},
            )
        )
        noop_plan.warnings.extend(skipped_warnings)
        return noop_plan
    remove_targets, blocks = _extract_moved_blocks(source_tree, expanded_names)
    rename_map = _active_rename(rename, moved_names)
    if rename_map:
        _assert_rename_targets_free(target_tree, rename_map)
        blocks = _apply_rename_to_blocks(blocks, rename_map)

    root = (
        Path(workspace_root)
        if workspace_root is not None
        else find_project_root(source_path)
    )
    import_resolution = _build_import_resolution(source_path, target_path, root)

    (
        new_source_tree,
        new_target_tree,
        imports_added,
        constants_added,
        shared_map,
        redundant_import_warnings,
    ) = _build_trees(
        source_tree,
        target_tree,
        blocks,
        remove_targets,
        shared_helpers,
        insert_after=insert_after,
        include_helpers=include_helpers,
        import_resolution=import_resolution,
        rename_map=rename_map,
    )

    if reexport:
        try:
            new_module_path = _module_path_from_file(target_path, root)
        except ValueError:
            new_module_path = target_path.stem
        new_source_tree = _inject_reexport(
            new_source_tree, new_module_path, moved_names
        )

    source_text_new, target_text_new = _render_and_validate(
        new_source_tree, new_target_tree
    )

    caller_texts, caller_rewrites, caller_warnings = _resolve_caller_phase(
        reexport, root, moved_names, source_path, target_path, rename_map
    )

    plan = _build_plan(
        MoveContext(
            source_text_new=source_text_new,
            target_text_new=target_text_new,
            moved_names=moved_names,
            imports_added=imports_added,
            constants_added=constants_added,
            shared_map=shared_map,
            callers_updated=caller_rewrites,
            redundant_import_warnings=redundant_import_warnings,
        )
    )
    plan.warnings.extend(skipped_warnings)
    plan.warnings.extend(caller_warnings)
    # Forward-refs to *renamed* symbols are rewritten in the moved code
    # (RenameSymbols.leave_Annotation); only moved-but-not-renamed names still
    # warrant the manual-update warning.
    unrenamed_moved = [n for n in moved_names if n not in rename_map]
    plan.warnings.extend(_string_forward_ref_warnings(source_tree, unrenamed_moved))
    deco_whitelist = SIDE_EFFECT_DECORATORS | (side_effect_decorators or frozenset())
    plan.warnings.extend(_side_effect_decorator_warnings(blocks, deco_whitelist))
    plan.warnings.extend(
        _fixture_scope_warnings(blocks, source_tree, source_path, target_path, root)
    )

    with _graph_cache_session():
        cycle = _cycle_check(
            root,
            source_path,
            target_path,
            new_source_tree,
            new_target_tree,
            caller_texts,
            blocks,
            source_tree,
        )
    _enforce_cycle(cycle, check, dry_run)

    if dry_run or check:
        return plan

    _apply_write(
        source_path,
        target_path,
        source_text,
        target_text,
        source_text_new,
        target_text_new,
        workspace_root,
        caller_texts,
    )
    plan.warnings.extend(_ruff_fix(source_path, target_path, reexport=reexport))
    return plan