Function request

st2common/st2common/services/workflows.py:222–313

(wf_def, ac_ex_db, st2_ctx, notify_cfg=None)

def request(wf_def, ac_ex_db, st2_ctx, notify_cfg=None):
    LOG.info("[%s] Processing action execution request for workflow.", str(ac_ex_db.id))

    # Load workflow definition into workflow spec model.
    spec_module = specs_loader.get_spec_module("native")
    wf_spec = spec_module.instantiate(wf_def)

    # Inspect the workflow spec.
    inspect(wf_spec, st2_ctx, raise_exception=True)

    # Identify the action to execute.
    action_db = action_utils.get_action_by_ref(ref=ac_ex_db.action["ref"])

    if not action_db:
        error = 'Unable to find action "%s".' % ac_ex_db.action["ref"]
        raise ac_exc.InvalidActionReferencedException(error)

    # Identify the runner for the action.
    runner_type_db = action_utils.get_runnertype_by_name(action_db.runner_type["name"])

    # Render action execution parameters.
    runner_params, action_params = param_utils.render_final_params(
        runner_type_db.runner_parameters,
        action_db.parameters,
        ac_ex_db.parameters,
        ac_ex_db.context,
    )

    # Instantiate the workflow conductor.
    conductor_params = {"inputs": action_params, "context": st2_ctx}
    conductor = conducting.WorkflowConductor(wf_spec, **conductor_params)

    # Serialize the conductor which initializes some internal values.
    data = conductor.serialize()

    # Create a record for workflow execution.
    wf_ex_db = wf_db_models.WorkflowExecutionDB(
        action_execution=str(ac_ex_db.id),
        spec=data["spec"],
        graph=data["graph"],
        input=data["input"],
        context=data["context"],
        state=data["state"],
        status=data["state"]["status"],
        output=data["output"],
        errors=data["errors"],
    )

    # Inspect that the list of tasks in the notify parameter exist in the workflow spec.
    if runner_params.get("notify"):
        invalid_tasks = list(
            set(runner_params.get("notify")) - set(wf_spec.tasks.keys())
        )

        if invalid_tasks:
            raise wf_exc.WorkflowExecutionException(
                "The following tasks in the notify parameter do not exist "
                "in the workflow definition: %s." % ", ".join(invalid_tasks)