def apply_parameters()

in jetstream/argo.py [0:0]

• 11 lines of code
• 8 McCabe index (conditional complexity)

def apply_parameters(manifest: dict[Any, Any], parameters: dict[str, Any]) -> dict[Any, Any]:
    """Apply custom parameters to the workflow manifest."""
    # Currently, there is no option for providing custom parameters for workflows.
    # apply_parameters works around this limitation by modifying the parsed manifest
    # and injecting custom parameters.
    workflow_parameters = manifest["spec"]["arguments"]["parameters"]
    for key, value in parameters.items():
        exists = False
        for workflow_param in workflow_parameters:
            # overwrite existing
            if workflow_param["name"] == key:
                # the array needs to be encoded as string
                # Argo doesn't support ' in it's configuration so replace with "
                workflow_param["value"] = str(value).replace("'", '"')
                exists = True

        if not exists:
            # append non-existing parameters
            workflow_parameters.append({"name": key, "value": value})

    return manifest