Class FanOutStep

src/specify_cli/workflows/steps/fan_out/__init__.py:11–74

Dispatch a step template for each item in a collection. The engine executes the nested ``step:`` template once per item, setting ``context.item`` for each iteration. Execution is currently sequential; ``max_concurrency`` is accepted but not enforced.

class FanOutStep(StepBase):
    """Dispatch a step template for each item in a collection.

    The engine executes the nested ``step:`` template once per item,
    setting ``context.item`` for each iteration.  Execution is
    currently sequential; ``max_concurrency`` is accepted but not
    enforced.
    """

    type_key = "fan-out"

    def execute(self, config: dict[str, Any], context: StepContext) -> StepResult:
        items_expr = config.get("items", "[]")
        items = evaluate_expression(items_expr, context)
        max_concurrency = config.get("max_concurrency", 1)
        step_template = config.get("step", {})

        if not isinstance(items, list):
            # A non-list here is a wiring error (the expression did not
            # resolve to a collection); silently fanning out over zero
            # items hides it. An explicit empty list remains valid input.
            return StepResult(
                status=StepStatus.FAILED,
                error=(
                    f"Fan-out step {config.get('id', '?')!r}: 'items' must "
                    f"resolve to a list, got {type(items).__name__} from "
                    f"{items_expr!r}."
                ),
                output={
                    "items": [],
                    "max_concurrency": max_concurrency,
                    "step_template": step_template,
                    "item_count": 0,
                },
            )

        return StepResult(
            status=StepStatus.COMPLETED,
            output={
                "items": items,
                "max_concurrency": max_concurrency,
                "step_template": step_template,
                "item_count": len(items),
            },
        )

    def validate(self, config: dict[str, Any]) -> list[str]:
        errors = super().validate(config)
        if "items" not in config:
            errors.append(
                f"Fan-out step {config.get('id', '?')!r} is missing "
                f"'items' field."
            )
        if "step" not in config:
            errors.append(
                f"Fan-out step {config.get('id', '?')!r} is missing "
                f"'step' field (nested step template)."
            )