Source code for cursus.steps.interfaces.io_view

"""
Per-step connection / I-O view (FZ 31e1d3d / 31e1d3b1 follow-up).

The Strategy+Facade collapse means a step is no longer a readable builder class — the
container source/destination paths, the runtime property-path references, and the (handler-derived)
nested training-channel expansion all live in the step's ``.step.yaml`` + its bound handler instead.
This module reads BOTH and renders one structured "what wires into / out of this step" view that the
``cursus steps io`` CLI and the ``steps.io`` MCP tool share.

It is the path/wiring analogue of ``catalog.step_spec`` (which gives the ports but NOT the container
paths or the channel fan-out): for each dependency it reports ``container_path`` (where the input
lands in the container) + the SageMaker training channel(s) it maps to; for each output it reports
``container_path`` (source) + ``property_path`` (the runtime ``properties.*`` reference a downstream
step resolves against). Pure introspection — no config, no SageMaker session.
"""

from __future__ import annotations

from typing import Any, Dict, List, Optional


def _channels_for(
    sagemaker_step_type: str,
    logical_name: str,
    container_path: Optional[str],
    declared_channels: Optional[List[str]],
) -> Optional[List[str]]:
    """The SageMaker training channel name(s) a dependency maps to, for Training steps only.

    Delegates to ``TrainingHandler.channels_for`` — the SINGLE SOURCE of the channel rule that
    ``TrainingHandler.get_inputs`` uses at build time, which itself prefers the channels declared
    in the ``.step.yaml`` — so this view can never drift from what the builder emits. Returns
    ``None`` for non-Training steps (no fan-out concept).
    """
    if sagemaker_step_type != "Training" or not container_path:
        return None
    from ...core.base.builder_templates import TrainingHandler

    return TrainingHandler.channels_for(logical_name, container_path, declared_channels)


[docs] def describe_step_io(step_name: str, job_type: Optional[str] = None) -> Dict[str, Any]: """Return the connection/I-O view for one step, resolved from its ``.step.yaml`` (+ job_type). Raises whatever ``load_interface`` raises for an unknown step (FileNotFoundError) — callers convert it to their own not-found envelope. """ from . import load_interface # lazy: avoids import cycle at module load from ...registry.step_names import get_sagemaker_step_type iface = load_interface(step_name, job_type=job_type) try: sm_type = get_sagemaker_step_type(step_name) except Exception: sm_type = None in_paths = iface.contract.expected_input_paths out_paths = iface.contract.expected_output_paths declared_channels = iface.contract.input_channels inputs: List[Dict[str, Any]] = [] for logical_name, dep in iface.spec.dependencies.items(): container_path = in_paths.get(logical_name) entry: Dict[str, Any] = { "logical_name": logical_name, "container_path": container_path, # where the input lands inside the container "required": dep.required, "type": getattr(getattr(dep, "dependency_type", None), "value", None) or str(getattr(dep, "dependency_type", "")), "compatible_sources": list(getattr(dep, "compatible_sources", []) or []), "semantic_keywords": list(getattr(dep, "semantic_keywords", []) or []), } channels = _channels_for( sm_type, logical_name, container_path, declared_channels.get(logical_name) ) if channels is not None: entry["channels"] = ( channels # nested SageMaker training channels (train/val/test, ...) ) inputs.append(entry) outputs: List[Dict[str, Any]] = [] for logical_name, out in iface.spec.outputs.items(): outputs.append( { "logical_name": logical_name, "container_path": out_paths.get( logical_name ), # source inside the container "property_path": getattr(out, "property_path", "") or "", # runtime reference "type": getattr(getattr(out, "output_type", None), "value", None) or str(getattr(out, "output_type", "")), "aliases": list(getattr(out, "aliases", []) or []), "data_type": getattr(out, "data_type", None), } ) return { "step_name": step_name, "step_type": iface.spec.step_type, "sagemaker_step_type": sm_type, "job_type": job_type, "inputs": inputs, "outputs": outputs, }
# Axes whose pattern the patterns-view reports + the builder method each maps to (for the # custom-override marker). Order = construction flow. _PATTERN_AXES = ( ("create_step", "create_step"), ("env_vars", "_get_environment_variables"), ("job_arguments", "_get_job_arguments"), ("inputs", "_get_inputs"), ("outputs", "_get_outputs"), ) # Compute-factory methods (any of these on a builder = a per-step compute override). Mapped to the # 'compute' axis in the patterns view. _COMPUTE_METHODS = ( "_create_processor", "_get_processor", "_create_estimator", "_create_model", "_create_transformer", )
[docs] def describe_step_patterns( step_name: str, job_type: Optional[str] = None ) -> Dict[str, Any]: """Return the per-axis PATTERN view for one step — the 'plugins' the TemplateStepBuilder uses (FZ 31e1d3j). Everything is DERIVED from the data that actually drives the build — the registry binding (``sagemaker_step_type`` + ``step_assembly`` → handler/verb) and the per-step ``.step.yaml`` contract DATA the handlers read (``env_vars``, ``job_arguments``, ``circular_ref_check``, ``skip_inputs``, ``input_source_overrides``, ``sink``, ``include_job_type_in_path``, ``source_dir``) — plus the output S3 prefix DERIVED from the step name (``canonical_to_snake(step_type)``). So this view CANNOT drift from behavior: there is no separate ``patterns:`` field — the pattern is the consequence of the data. Where a builder still hand-overrides a method (a genuine per-step deviation), that axis is marked ``custom_override`` so the user sees exactly where the step departs from the declarative patterns. """ from . import load_interface # lazy: avoids import cycle at module load from ...registry.step_names import get_sagemaker_step_type iface = load_interface(step_name, job_type=job_type) c = iface.contract try: sm_type = get_sagemaker_step_type(step_name) except Exception: sm_type = None # Fall back to the .step.yaml registry block (now a real section) when the registry lookup is # unavailable (e.g. offline) — the YAML is the same source of truth and keeps the view populated. if not sm_type: sm_type = getattr(getattr(iface, "registry", None), "sagemaker_step_type", None) # --- which builder methods does this step still hand-override? (genuine keeps) --- # Two-tier so the answer is faithful in every state (the per-step builder files are gone — builders # are SYNTHESIZED fileless shells, so most steps have NO overrides by definition): # 1. load the class and inspect __dict__ (most accurate — incl. a synthesized shell, which has # no own methods, so overridden = {} correctly); # 2. if the class won't load (e.g. SAIS SDK absent offline), it is a pure declarative shell we # can't introspect here → report no overrides and attach a note so a caller never mistakes a # "couldn't check" for a genuine "no override". overridden: set = set() builder_note = None method_names = [m for _, m in _PATTERN_AXES] + list(_COMPUTE_METHODS) builder_cls = None try: from ...step_catalog.step_catalog import StepCatalog builder_cls = StepCatalog().load_builder_class(step_name) except Exception: builder_cls = None if builder_cls is not None: overridden = { m for m in method_names if m in getattr(builder_cls, "__dict__", {}) } else: builder_note = ( "declarative shell — builder synthesized at runtime and not importable here " "(e.g. SAIS SDK absent); custom_override not introspected, reported as no overrides" ) # --- create_step: the bound construction handler (the registry binding) --- reg = getattr(iface, "registry", None) # step_assembly is now declared in patterns: (the blueprint axis read by _auto_bind_handler); # prefer it so this view binds from the SAME source as the build (closes the FZ 31e1d3f drift), # falling back to the legacy registry.step_assembly for any not-yet-migrated YAML. pat = getattr(iface, "patterns", None) step_assembly = getattr(pat, "step_assembly", None) or getattr( reg, "step_assembly", None ) create_step_requires = getattr(reg, "requires", "none") handler_name = None try: from ...core.base.builder_templates import resolve_handler h = ( resolve_handler(sm_type, step_assembly) if sm_type == "Processing" else resolve_handler(sm_type) ) handler_name = type(h).__name__ except Exception: handler_name = None patterns: Dict[str, Any] = {} patterns["create_step"] = { "handler": handler_name, "sagemaker_step_type": sm_type, "step_assembly": step_assembly, # dependency axis: the BUILD-time 3rd-party pkg the create_step pattern needs (FZ 31e1d3l). # "secure_ai_sandbox_workflow_python_sdk" = a hard module-level SAIS import (SDKDelegation # steps); "none" = native (sagemaker-only). Declared in .step.yaml registry.requires. "requires": create_step_requires, "custom_override": "create_step" in overridden, } # --- compute: the declared SDK processor/estimator the template builds (FZ 31e1d3k) --- # compute is a TOP-LEVEL .step.yaml section now; read it off the interface (falls back to the # contract mirror for any caller that passes only a contract). cspec = getattr(iface, "compute", None) or getattr(c, "compute", None) compute_kind = getattr(cspec, "kind", None) # map kind -> the SDK class surfaced to the user _kind_to_class = { "sklearn": "SKLearnProcessor", "xgboost": "XGBoostProcessor", "script": "ScriptProcessor", "framework": "FrameworkProcessor", "estimator": getattr(cspec, "sdk_class", None), "model": getattr(cspec, "sdk_class", None), "transformer": "Transformer", } compute_override = ( "_create_processor" in overridden or "_get_processor" in overridden or "_create_estimator" in overridden or "_create_model" in overridden or "_create_transformer" in overridden ) compute_requires = getattr(cspec, "requires", "none") patterns["compute"] = { "kind": compute_kind, "sdk_class": _kind_to_class.get(compute_kind) if compute_kind else None, "framework_version_field": getattr(cspec, "framework_version_field", None), "lock_training_region": getattr(cspec, "lock_training_region", False), # dependency axis: the BUILD-time 3rd-party pkg this compute pattern needs (FZ 31e1d3l). # "mods_workflow_core" = the script/kms_network ScriptProcessor path (lazy import, no # fallback); "none" = sagemaker-only. Declared in .step.yaml contract.compute.requires. "requires": compute_requires, "source": "config-built by builder_base._create_compute() from the compute descriptor" if compute_kind else "per-step factory (no compute descriptor)", "custom_override": compute_override, } # --- env vars: declared names (config supplies values) --- patterns["env_vars"] = { "source": "config.get_environment_variables() (config is the value source)", "declared_required": list(getattr(c, "required_env_vars", []) or []), "declared_optional": sorted((getattr(c, "optional_env_vars", {}) or {}).keys()), "custom_override": "_get_environment_variables" in overridden, } # --- job arguments: declarative flag→source --- patterns["job_arguments"] = { "source": "config.get_job_arguments() (config is the value source)", "declared": [ {"flag": d.flag, "source": d.source} for d in getattr(c, "job_arguments", []) ], "custom_override": "_get_job_arguments" in overridden, } # --- inputs: the standard loop + which deviation flags are active --- in_devs = [] if getattr(c, "circular_ref_check", False): in_devs.append("circular_ref_check") if getattr(c, "skip_inputs", None): in_devs.append(f"skip_inputs={list(c.skip_inputs)}") if getattr(c, "input_source_overrides", None): in_devs.append(f"input_source_overrides={dict(c.input_source_overrides)}") if getattr(c, "source_dir", False): in_devs.append("source_dir (split entry_point + source_dir)") patterns["inputs"] = { "pattern": "standard spec×contract loop" + (" + " + ", ".join(in_devs) if in_devs else ""), "deviations": in_devs, "custom_override": "_get_inputs" in overridden, } # --- outputs: sink vs generated-destination token/job_type --- # The S3 prefix token is DERIVED from the step name (canonical_to_snake) by default, unless the # contract sets an OPT-IN output_path_token override (FZ 31e1d3f1b re-introduced, default-off). # include_job_type_in_path remains a per-step knob. if getattr(c, "sink", False): out_pattern = "sink (no outputs)" _token = None _token_source = None else: from ...step_catalog.naming import canonical_to_snake _override = getattr(c, "output_path_token", None) _token = _override or canonical_to_snake(iface.spec.step_type) _token_source = ( "override: contract.output_path_token" if _override else "derived: canonical_to_snake(step_type)" ) token = _token jt = getattr(c, "include_job_type_in_path", True) out_pattern = f"generated destination Join(base, {token}{', job_type' if jt else ''}, logical_name)" patterns["outputs"] = { "pattern": out_pattern, "output_path_token": _token, "output_path_token_source": _token_source, "include_job_type_in_path": getattr(c, "include_job_type_in_path", True), "sink": getattr(c, "sink", False), "custom_override": "_get_outputs" in overridden, } # --- dependency-axis rollup: the step's whole 3rd-party footprint at a glance (FZ 31e1d3l), # BUILD-time (create_step / compute) kept separate from RUNTIME (the container script). This is # the mods-vs-native split the user asked to surface — sourced entirely from .step.yaml DATA. --- runtime_requires = list(getattr(c, "runtime_requires", []) or []) build_time = {} if create_step_requires and create_step_requires != "none": build_time["create_step"] = create_step_requires if compute_requires and compute_requires != "none": build_time["compute"] = compute_requires result: Dict[str, Any] = { "step_name": step_name, "step_type": iface.spec.step_type, "sagemaker_step_type": sm_type, "job_type": job_type, "patterns": patterns, "dependencies": { "build_time": build_time, # {axis: pkg} — absent axes are native (sagemaker-only) "runtime": runtime_requires, # pkgs the script needs in-container (NOT at build time) "native": not build_time and not runtime_requires, }, } if builder_note: result["note"] = builder_note return result