Skip to content

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
Python
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}")

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
Python
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
Python
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
Python
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
Python
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
Python
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
Python
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
Python
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
    )
    remove_targets, blocks = _extract_moved_blocks(source_tree, expanded_names)
    rename_map = _active_rename(rename, moved_names)
    if rename_map:
        blocks = _apply_rename_to_blocks(blocks, rename_map)

    root = (
        Path(workspace_root)
        if workspace_root is not None
        else _find_workspace_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 = _resolve_caller_phase(
        reexport, root, moved_names, source_path, target_path, rename_map
    )

    plan = _build_plan(
        source_text_new,
        target_text_new,
        moved_names,
        imports_added,
        constants_added,
        shared_map,
        callers_updated=caller_rewrites,
        redundant_import_warnings=redundant_import_warnings,
    )
    plan.warnings.extend(skipped_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)
    )

    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