rs_workflows/payload_builder.md

<< Back to index

.

TaskTableError

Bases: ValueError

Errors related to Task Table parsing/validation.

Source code in docs/rs-client-libraries/rs_workflows/payload_builder.py

class TaskTableError(ValueError):
    """Errors related to Task Table parsing/validation."""

build_cql2_json(query, values)

Recursively replaces placeholders of the form {var} in a dictionary or list using the mapping from 'values'.

Source code in docs/rs-client-libraries/rs_workflows/payload_builder.py

def build_cql2_json(query: dict[str, Any], values: dict[str, Any]):
    """
    Recursively replaces placeholders of the form {var} in a dictionary or list
    using the mapping from 'values'.
    """

    def _replace(item):
        if isinstance(item, str):
            # replace if found, else keep
            match = VARIABLE_PATTERN.match(item)
            return values.get(match.group(1), item) if match else item
        if isinstance(item, list):
            return [_replace(x) for x in item]
        if isinstance(item, dict):
            return {k: _replace(v) for k, v in item.items()}
        return item

    # Work on a deep copy so we don't mutate the original
    return _replace(deepcopy(query))["stac"]

build_unit_list(tasktable, pipeline=None, unit=None, processing_mode=None)

Build the list of units needed for the payload, from the tasktable given and the flow mode (pipeline or unit). If the mode is "unit", the list returned will only contain the requested unit, with its input and output linked to the external input and output. If the mode is "pipeline", the list will contain all the ordered units of the pipeline, with their inputs and outputs correctly chained if needed.

Parameters:

Name	Type	Description	Default
tasktable	dict[str, Any]	dictionary containing the content of the tasktable	required
pipeline	str | None	name of the pipeline to build, if the mode is "pipeline". If "unit" is given, has to be None.	None
unit	str | None	name of the unit to build, if the mode is "unit". If "pipeline" is given, has to be None.	None
processing_mode	Iterable[str] | None	list of processing modes used for this flow. Optional	None

Source code in docs/rs-client-libraries/rs_workflows/payload_builder.py

def build_unit_list(
    tasktable: dict[str, Any],
    pipeline: str | None = None,
    unit: str | None = None,
    processing_mode: Iterable[str] | None = None,
) -> list[dict[str, Any]]:
    """
    Build the list of units needed for the payload, from the tasktable given and the flow mode (pipeline or unit).
    If the mode is "unit", the list returned will only contain the requested unit, with its input and output linked
    to the external input and output.
    If the mode is "pipeline", the list will contain all the ordered units of the pipeline, with their inputs and
    outputs correctly chained if needed.

    Args:
        tasktable: dictionary containing the content of the tasktable
        pipeline: name of the pipeline to build, if the mode is "pipeline". If "unit" is given, has to be None.
        unit: name of the unit to build, if the mode is "unit". If "pipeline" is given, has to be None.
        processing_mode: list of processing modes used for this flow. Optional
    """
    io = tasktable.get("io") if isinstance(tasktable, dict) else None
    get_logger(__name__).debug(
        "Task table IO diagnostic: file=%s expected_io_type=dict tasktable_type=%s io_type=%s io_keys=%s "
        "pipeline=%s unit=%s",
        __file__,
        type(tasktable).__name__,
        type(io).__name__,
        list(io) if isinstance(io, dict) else None,
        pipeline,
        unit,
    )

    # Validate pipelines shape
    if not isinstance(tasktable, dict):
        raise TaskTableError(f"Task table root must be a JSON object (dict): {tasktable}")
    if "pipelines" not in tasktable or not isinstance(tasktable["pipelines"], list):
        raise TaskTableError(f"Missing or invalid 'pipelines' list in task table: {tasktable}")
    if "units" not in tasktable or not isinstance(tasktable["units"], list):
        raise TaskTableError(f"Missing or invalid 'units' list in task table: {tasktable}")
    if "io" not in tasktable or not isinstance(tasktable["io"], dict):
        raise TaskTableError(f"Missing or invalid 'io' dict in task table: {tasktable}")

    if pipeline and unit:
        raise TaskTableError("Provide either 'pipeline' or 'unit', not both.")
    if not pipeline and not unit:
        raise TaskTableError("One of 'pipeline' or 'unit' must be provided.")

    # Retrieve list of units from "units" field
    units_index: dict[str, dict[str, Any]] = {}
    for u in tasktable["units"]:
        if isinstance(u, dict) and isinstance(u.get("name"), str):
            units_index[u["name"]] = u
    if not units_index:
        raise TaskTableError('No valid unit entries found in "units".')

    # Retrieve details of inputs/adfs/outputs from "io" field
    io_index: dict[str, dict[str, dict[str, Any]]] = {}
    for io_type in ("input", "adfs", "output"):
        io_index[io_type] = {entry["name"]: entry for entry in get_io_entries(tasktable, io_type)}

    out_units: list[dict[str, Any]] = []

    if unit:
        # Handle only one unit
        unit_details = _build_single_unit_details(
            unit,
            units_index,
            io_index,
            processing_mode,
        )
        out_units.append(unit_details)

    if pipeline:
        # Retrieve pipeline definition from the list of pipelines
        full_pipeline = next((p for p in tasktable.get("pipelines", []) if p.get("name") == pipeline), None)

        if not full_pipeline:
            raise TaskTableError(
                f"Could not find pipeline named '{pipeline}' in tasktable. "
                f"Existing pipelines: {tasktable.get("pipelines", [])}",
            )

        # Retrieve list of steps from the pipeline
        pipeline_steps = full_pipeline.get("steps")
        if not pipeline_steps or not isinstance(pipeline_steps, list):
            raise TaskTableError(f"Pipeline '{pipeline}' is missing a valid 'steps' list.")

        for step in pipeline_steps:
            # Validate step structure
            if not isinstance(step, dict):
                raise TaskTableError(
                    f"Step in pipeline '{pipeline}' doesn't have the expected format (expected JSON dict): {step}",
                )
            for field in ("unit_name", "step_id", "input_products", "output_products"):
                if field not in step:
                    raise TaskTableError(f"Step in pipeline '{pipeline}' is missing mandatory field '{field}': {step}")

            # Build unit details for this step
            unit_details = _build_single_unit_details(
                step["unit_name"],
                units_index,
                io_index,
                processing_mode,
                full_pipeline=full_pipeline,
                step_id=step["step_id"],
                parameters=step.get("parameters", None),
            )

            out_units.append(unit_details)

    return out_units

get_io_entries(tasktable, role)

Read grouped or product-name-indexed I/O without changing the task table.

Source code in docs/rs-client-libraries/rs_workflows/payload_builder.py

def get_io_entries(tasktable: dict[str, Any], role: str) -> list[dict[str, Any]]:
    """Read grouped or product-name-indexed I/O without changing the task table."""
    io = tasktable["io"]
    if not isinstance(io, dict):
        raise TaskTableError("Task table 'io' must be a dictionary")
    if any(isinstance(io.get(key), list) for key in ("input", "adfs", "output")):
        return io.get(role, [])
    entries = []
    for name, details in io.items():
        if not isinstance(details, dict):
            raise TaskTableError(f"Invalid IO definition for '{name}': expected an object")
        if details.get("name", name) != name:
            raise TaskTableError(f"IO key '{name}' conflicts with name '{details['name']}'")
        entries.append({**details, "name": name})
    return entries

replace_external_variables(obj, external_variables)

Replace configured external variables with the input parameters

Source code in docs/rs-client-libraries/rs_workflows/payload_builder.py

def replace_external_variables(obj, external_variables: dict[str, Any] | None):
    """Replace configured external variables with the input parameters"""
    if isinstance(obj, dict):
        return {k: replace_external_variables(v, external_variables) for k, v in obj.items()}
    if isinstance(obj, list):
        return [replace_external_variables(v, external_variables) for v in obj]
    if isinstance(obj, str) and external_variables:

        def replacer(match: re.Match):
            variable_name = match.group(1)
            if variable_name not in external_variables:
                raise TaskTableError(f"Unknown external variable: {variable_name}")
            value = external_variables[variable_name]
            if "datetime" in variable_name:
                if value is None:
                    raise TaskTableError(
                        f"External variable '{variable_name}' is required by the task table but was not provided",
                    )
                return strftime_millis(value)
            return str(value)

        return EXTERNAL_VAR_PATTERN.sub(replacer, obj)
    return obj