Quelle  taskgraph.py   Sprache: Python

# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

"""
Tools for interacting with existing taskgraphs.
"""

from taskgraph.util.taskcluster import find_task_id, get_artifact


def find_decision_task(parameters, graph_config):
    """Given the parameters for this action, find the taskId of the decision
    task"""
    head_rev_param = "{}head_rev".format(graph_config["project-repo-param-prefix"])
    return find_task_id(
        "{}.v2.{}.revision.{}.taskgraph.decision".format(
            graph_config["trust-domain"],
            parameters["project"],
            parameters[head_rev_param],
        )
    )


def find_existing_tasks(previous_graph_ids):
    existing_tasks = {}
    for previous_graph_id in previous_graph_ids:
        label_to_taskid = get_artifact(previous_graph_id, "public/label-to-taskid.json")
        existing_tasks.update(label_to_taskid)
    return existing_tasks


def find_existing_tasks_from_previous_kinds(
    full_task_graph, previous_graph_ids, rebuild_kinds
):
    """Given a list of previous decision/action taskIds and kinds to ignore
    from the previous graphs, return a dictionary of labels-to-taskids to use
    as ``existing_tasks`` in the optimization step."""
    existing_tasks = find_existing_tasks(previous_graph_ids)
    kind_labels = {
        t.label
        for t in full_task_graph.tasks.values()
        if t.attributes["kind"] not in rebuild_kinds
    }
    return {
        label: taskid
        for (label, taskid) in existing_tasks.items()
        if label in kind_labels
    }