Skip to content

Protocol scaffolder

protocol_scaffolder

ProtocolScaffoldRequest dataclass

Hold one validated protocol-profile request.

Source code in packages/axm-init/src/axm_init/core/protocol_scaffolder.py
Python
@dataclass(frozen=True, slots=True)
class ProtocolScaffoldRequest:
    """Hold one validated protocol-profile request."""

    profile: str
    domain: str
    unit: str | None
    protocols: tuple[ProtocolScaffoldDecl, ...]
    preview: bool

build_protocol_scaffold_result(plan, *, profile, mode, root, graph_names, preview)

Project a typed planner result into the shared structured result.

Source code in packages/axm-init/src/axm_init/core/protocol_scaffolder.py
Python
def build_protocol_scaffold_result(  # noqa: PLR0913
    plan: ProtocolScaffoldPlan,
    *,
    profile: str,
    mode: str,
    root: PurePosixPath,
    graph_names: tuple[str, ...],
    preview: bool,
) -> ScaffoldResult:
    """Project a typed planner result into the shared structured result."""
    grouped: dict[PlanStatus, list[str]] = {status: [] for status in PlanStatus}
    for operation in plan.operations:
        grouped[operation.status].append(operation.path.as_posix())
    return ScaffoldResult(
        success=True,
        path=root.as_posix(),
        message="Protocol scaffold preview",
        profile=profile,
        mode=mode,
        root=root.as_posix(),
        preview=preview,
        created=grouped[PlanStatus.CREATE],
        updated=grouped[PlanStatus.UPDATE],
        unchanged=grouped[PlanStatus.UNCHANGED],
        conflicts=grouped[PlanStatus.CONFLICT],
        protocols=list(graph_names),
    )

prepare_protocol_request(*, profile, domain, unit, protocols, preview, framework)

Validate protocol options before any target filesystem mutation.

Source code in packages/axm-init/src/axm_init/core/protocol_scaffolder.py
Python
def prepare_protocol_request(  # noqa: PLR0913
    *,
    profile: str | None,
    domain: str | None,
    unit: str | None,
    protocols: list[ProtocolPayload] | str | None,
    preview: bool,
    framework: Framework,
) -> ProtocolScaffoldRequest | str | None:
    """Validate protocol options before any target filesystem mutation."""
    requested = any((profile, domain, unit, protocols is not None, preview))
    if not requested:
        return None
    try:
        valid_profile, valid_domain = _validate_options(profile, domain, framework)
        payloads = _payloads(protocols)
        _validate_payloads(payloads, preview=preview, unit=unit)
        declarations = tuple(
            ProtocolScaffoldDecl.model_validate(
                {**payload, "domain": valid_domain, "unit": unit}
            )
            for payload in payloads
        )
        _validate_declarations(
            declarations,
            domain=valid_domain,
            unit=unit,
        )
    except (TypeError, ValueError, json.JSONDecodeError) as exc:
        return str(exc)
    return ProtocolScaffoldRequest(
        profile=valid_profile,
        domain=valid_domain,
        unit=unit,
        protocols=declarations,
        preview=preview,
    )

preview_protocol_scaffold(root, request)

Plan declarations and serialize applications by canonical target root.

Source code in packages/axm-init/src/axm_init/core/protocol_scaffolder.py
Python
def preview_protocol_scaffold(
    root: Path,
    request: ProtocolScaffoldRequest,
) -> ScaffoldResult:
    """Plan declarations and serialize applications by canonical target root."""
    if request.preview:
        return _preview_protocol_scaffold_unlocked(root, request)
    canonical_root = root.resolve()
    with _target_root_lock(canonical_root):
        return _preview_protocol_scaffold_unlocked(canonical_root, request)

register_protocol_profile(root, domain)

Persist the profile while serialized with applications on this root.

Source code in packages/axm-init/src/axm_init/core/protocol_scaffolder.py
Python
def register_protocol_profile(root: Path, domain: str) -> None:
    """Persist the profile while serialized with applications on this root."""
    canonical_root = root.resolve()
    with _target_root_lock(canonical_root):
        _register_protocol_profile_unlocked(canonical_root, domain)