api/libs/workflow_template_values.py

Rama main · réplica de lectura del repositorio original.

from collections.abc import Iterator, Sequence
from typing import Protocol

from graphon.nodes.base.variable_template_parser import extract_selectors_from_template
from graphon.variables.segments import Segment
from graphon.variables.template_resolution import VARIABLE_PATTERN

type TemplateValue = str | Sequence[str] | int | float | None


class TemplateVariablePool(Protocol):
    def get(self, selector: Sequence[str], /) -> Segment | None: ...


def template_value_selectors(value: TemplateValue) -> Iterator[list[str]]:
    """Extract workflow references from a scalar string or string sequence.

    Preconditions: ``value`` is a validated metadata condition value.
    Postconditions: Yield selectors without changing the value; constants
    yield nothing.
    """
    values = [value] if isinstance(value, str) else value
    if isinstance(values, Sequence):
        for template in values:
            for reference in extract_selectors_from_template(template):
                yield list(reference.value_selector)


def resolve_template_value(value: TemplateValue, variable_pool: TemplateVariablePool) -> TemplateValue:
    """Resolve scalar values and string sequences against a workflow variable pool.

    Preconditions: ``value`` is a validated metadata condition value and the
    pool belongs to the current run.
    Postconditions: Return the resolved value without changing either input.
    A whole scalar reference retains its string or numeric type;
    interpolated text and elements of string lists remain strings. Missing
    references fail explicitly rather than becoming literal selector text.
    """

    def get_segment(selector: str) -> Segment:
        """Return an existing dot-separated selector or raise ``ValueError``."""
        segment = variable_pool.get(selector.split("."))
        if segment is None:
            raise ValueError(f"Workflow variable {selector} is not available.")
        return segment

    def resolve_template(template: str) -> str | int | float | None:
        """Resolve a string template without changing the pool; fail on missing refs."""
        if match := VARIABLE_PATTERN.fullmatch(template):
            resolved = get_segment(match.group(1)).to_object()
            if resolved is None or isinstance(resolved, str | float):
                return resolved
            if isinstance(resolved, int) and not isinstance(resolved, bool):
                return resolved
            return str(resolved)
        return VARIABLE_PATTERN.sub(lambda match: get_segment(match.group(1)).text, template)

    if isinstance(value, str):
        return resolve_template(value)
    if isinstance(value, Sequence):
        return [str(resolve_template(item)) for item in value]
    return value