Copier

copier

Copier adapter for template-based scaffolding.

CopierAdapter

Adapter for Copier template operations.

Wraps Copier's run_copy function with a Pydantic-based interface and returns structured ScaffoldResult.

Source code in packages/axm-init/src/axm_init/adapters/copier.py

class CopierAdapter:
    """Adapter for Copier template operations.

    Wraps Copier's run_copy function with a Pydantic-based interface
    and returns structured ScaffoldResult.
    """

    @staticmethod
    def _do_copy(config: CopierConfig) -> None:
        """Run copier, offloading to a thread if an event loop is active.

        Copier (via prompt_toolkit) calls ``asyncio.run()`` internally.
        When we are already inside an async event loop (e.g. MCP server),
        this raises ``RuntimeError: asyncio.run() cannot be called from
        a running event loop``.  The fix: detect the running loop and
        execute the blocking copy in a **separate thread** which gets
        its own event loop context.
        """
        import asyncio

        def _run() -> None:
            # ``run_copy`` declares ``data: dict[str, Any] | None``;
            # converting our ``Mapping[str, object]`` to a plain ``dict``
            # widens cleanly to the expected type.
            run_copy(
                src_path=str(config.template_path),
                dst_path=config.destination,
                data=dict(config.data),
                defaults=config.defaults,
                overwrite=config.overwrite,
                unsafe=config.trust_template,
                skip_tasks=config.skip_tasks,
                answers_file=config.answers_file,
                exclude=config.exclude,
            )

        try:
            asyncio.get_running_loop()
        except RuntimeError:
            # No event loop — safe to call directly (CLI context).
            _run()
        else:
            # Inside an event loop (MCP server) — offload without blocking
            # the running loop on a synchronous ``future.result()``.
            _offload_to_thread(_run)

    def apply_chain(
        self,
        layers: list[TemplateLayer],
        destination: Path,
        data: Mapping[str, object],
        *,
        record_answers: bool = True,
    ) -> ScaffoldResult:
        """Apply ordered template layers to one destination.

        Each layer receives caller data overlaid with its own data and keeps a
        dedicated answers file so Copier can reapply its ownership rules
        independently from the other layers.

        With ``record_answers=False``, exclude the engine's answer destinations
        before rendering and omit fallback answer creation. Existing answers
        stay untouched, including answers belonging to the same layer.
        """
        result = ScaffoldResult(
            success=True,
            path=str(destination),
            message="No template layers to apply",
        )
        for layer in layers:
            layer_data = dict(data)
            layer_data.update(layer.data)
            answers_file = Path(f".copier-answers.{layer.name}.yml")
            layer_data["_src_path"] = str(layer.path)
            layer_data["_answers_file"] = str(answers_file)
            result = self.copy(
                CopierConfig(
                    template_path=layer.path,
                    destination=destination,
                    data=layer_data,
                    overwrite=True,
                    trust_template=True,
                    answers_file=answers_file,
                    exclude=()
                    if record_answers
                    else (f"/{answers_file}", "/.copier-answers.yml"),
                )
            )
            if not result.success:
                return result
            answers_path = destination / answers_file
            if record_answers and not answers_path.exists():
                answers_path.write_text(
                    json.dumps({"_src_path": str(layer.path)}, indent=2) + "\n"
                )
        return result

    def copy(self, config: CopierConfig) -> ScaffoldResult:
        """Execute Copier copy operation.

        Suppresses stdout/stderr via the scoped :func:`_suppress_output`
        context manager so that post-copy tasks (git init, uv sync,
        pre-commit install) don't pollute the parent process stdio —
        critical when running inside an MCP server.  Suppression is scoped
        to the interpreter-level streams and never mutates the process-global
        file descriptors 1/2, so concurrent writers on those fds are
        unaffected.

        Args:
            config: Copier configuration with template path, destination, and data.

        Returns:
            ScaffoldResult with success status and path.
        """
        if config.trust_template:
            logger.warning(
                "Running Copier with unsafe=True — template may execute "
                "arbitrary post-copy tasks."
            )
        try:
            with _suppress_output():
                self._do_copy(config)
            # Walk destination to collect created files, excluding noise
            # from post-copy tasks (.git, .venv, __pycache__, node_modules).
            _excluded = {
                ".git",
                ".venv",
                "__pycache__",
                "node_modules",
                ".mypy_cache",
            }
            created: list[str] = sorted(
                str(p.relative_to(config.destination))
                for p in config.destination.rglob("*")
                if p.is_file()
                and not any(
                    part in _excluded
                    for part in p.relative_to(config.destination).parts[:-1]
                )
            )
            return ScaffoldResult(
                success=True,
                path=str(config.destination),
                message="Project scaffolded via Copier",
                files_created=created,
            )
        except Exception as e:
            return ScaffoldResult(
                success=False,
                path=str(config.destination),
                message=f"Copier failed: {e}",
            )

apply_chain(layers, destination, data, *, record_answers=True)

Apply ordered template layers to one destination.

Each layer receives caller data overlaid with its own data and keeps a dedicated answers file so Copier can reapply its ownership rules independently from the other layers.

With record_answers=False, exclude the engine's answer destinations before rendering and omit fallback answer creation. Existing answers stay untouched, including answers belonging to the same layer.

Source code in packages/axm-init/src/axm_init/adapters/copier.py

def apply_chain(
    self,
    layers: list[TemplateLayer],
    destination: Path,
    data: Mapping[str, object],
    *,
    record_answers: bool = True,
) -> ScaffoldResult:
    """Apply ordered template layers to one destination.

    Each layer receives caller data overlaid with its own data and keeps a
    dedicated answers file so Copier can reapply its ownership rules
    independently from the other layers.

    With ``record_answers=False``, exclude the engine's answer destinations
    before rendering and omit fallback answer creation. Existing answers
    stay untouched, including answers belonging to the same layer.
    """
    result = ScaffoldResult(
        success=True,
        path=str(destination),
        message="No template layers to apply",
    )
    for layer in layers:
        layer_data = dict(data)
        layer_data.update(layer.data)
        answers_file = Path(f".copier-answers.{layer.name}.yml")
        layer_data["_src_path"] = str(layer.path)
        layer_data["_answers_file"] = str(answers_file)
        result = self.copy(
            CopierConfig(
                template_path=layer.path,
                destination=destination,
                data=layer_data,
                overwrite=True,
                trust_template=True,
                answers_file=answers_file,
                exclude=()
                if record_answers
                else (f"/{answers_file}", "/.copier-answers.yml"),
            )
        )
        if not result.success:
            return result
        answers_path = destination / answers_file
        if record_answers and not answers_path.exists():
            answers_path.write_text(
                json.dumps({"_src_path": str(layer.path)}, indent=2) + "\n"
            )
    return result

copy(config)

Execute Copier copy operation.

Suppresses stdout/stderr via the scoped :func:_suppress_output context manager so that post-copy tasks (git init, uv sync, pre-commit install) don't pollute the parent process stdio — critical when running inside an MCP server. Suppression is scoped to the interpreter-level streams and never mutates the process-global file descriptors 1/2, so concurrent writers on those fds are unaffected.

Parameters:

Name	Type	Description	Default
config	CopierConfig	Copier configuration with template path, destination, and data.	required

Returns:

Type	Description
ScaffoldResult	ScaffoldResult with success status and path.

Source code in packages/axm-init/src/axm_init/adapters/copier.py

def copy(self, config: CopierConfig) -> ScaffoldResult:
    """Execute Copier copy operation.

    Suppresses stdout/stderr via the scoped :func:`_suppress_output`
    context manager so that post-copy tasks (git init, uv sync,
    pre-commit install) don't pollute the parent process stdio —
    critical when running inside an MCP server.  Suppression is scoped
    to the interpreter-level streams and never mutates the process-global
    file descriptors 1/2, so concurrent writers on those fds are
    unaffected.

    Args:
        config: Copier configuration with template path, destination, and data.

    Returns:
        ScaffoldResult with success status and path.
    """
    if config.trust_template:
        logger.warning(
            "Running Copier with unsafe=True — template may execute "
            "arbitrary post-copy tasks."
        )
    try:
        with _suppress_output():
            self._do_copy(config)
        # Walk destination to collect created files, excluding noise
        # from post-copy tasks (.git, .venv, __pycache__, node_modules).
        _excluded = {
            ".git",
            ".venv",
            "__pycache__",
            "node_modules",
            ".mypy_cache",
        }
        created: list[str] = sorted(
            str(p.relative_to(config.destination))
            for p in config.destination.rglob("*")
            if p.is_file()
            and not any(
                part in _excluded
                for part in p.relative_to(config.destination).parts[:-1]
            )
        )
        return ScaffoldResult(
            success=True,
            path=str(config.destination),
            message="Project scaffolded via Copier",
            files_created=created,
        )
    except Exception as e:
        return ScaffoldResult(
            success=False,
            path=str(config.destination),
            message=f"Copier failed: {e}",
        )

CopierConfig

Bases: BaseModel

Configuration for Copier execution.

Note: type: ignore[explicit-any] flags pydantic BaseModel internals (third-party).

Source code in packages/axm-init/src/axm_init/adapters/copier.py

class CopierConfig(BaseModel):  # type: ignore[explicit-any]
    """Configuration for Copier execution.

    Note: ``type: ignore[explicit-any]`` flags pydantic ``BaseModel``
    internals (third-party).
    """

    template_path: Path
    destination: Path
    data: Mapping[str, object]
    defaults: bool = True
    overwrite: bool = False
    trust_template: bool = False
    skip_tasks: bool = False
    answers_file: Path | None = None
    """Render the template without running its ``_tasks``.

    The bundled templates declare post-copy tasks that shell out to ``git init``
    and two ``uv add`` invocations, so a single render resolves 72 packages and
    installs 72 of them — measured at 2.23s and 239 MB against 0.60s and 0.1 MB
    with tasks skipped. The rendered tree is identical either way: only the
    tasks' side-effects are absent, and the deterministic ones (LICENSE,
    ``.python-version``, ``uv.lock``, the pre-commit hook) are cheap to
    synthesize. Production scaffolding leaves this ``False``; a caller that only
    needs the rendered tree — a test asserting template structure, say — sets it
    to ``True`` rather than paying for a package installation it never inspects.
    """

    exclude: tuple[str, ...] = ()
    """Destination-relative patterns omitted by Copier during rendering."""

    model_config = ConfigDict(extra="forbid")

answers_file = None class-attribute instance-attribute

Render the template without running its _tasks.

The bundled templates declare post-copy tasks that shell out to git init and two uv add invocations, so a single render resolves 72 packages and installs 72 of them — measured at 2.23s and 239 MB against 0.60s and 0.1 MB with tasks skipped. The rendered tree is identical either way: only the tasks' side-effects are absent, and the deterministic ones (LICENSE, .python-version, uv.lock, the pre-commit hook) are cheap to synthesize. Production scaffolding leaves this False; a caller that only needs the rendered tree — a test asserting template structure, say — sets it to True rather than paying for a package installation it never inspects.

exclude = () class-attribute instance-attribute

Destination-relative patterns omitted by Copier during rendering.