Render Dataset Conditions in DAG Graph view

[bbovenzi]

Before we were just rendering a json object of the any/all conditions of dataset events for a dag to run. Now, we interpret that and render it in the graph view with logical gates.

Datasets that actually had events will be highlighted with a different border so it's easy to see what triggered the selected run.

Also, added a check to still create a dataset node if there is a dataset event even if the getDatasets endpoint didnt return anything.

These are both workarounds. It would best to refactor the dag graph python code to accept a with_datasets param and handle this logic and include dataset aliases too. Hopefully, I can make that a follow-up PR.

---

^ Add meaningful description above
Read the Pull Request Guidelines for more information.
In case of fundamental code changes, an Airflow Improvement Proposal (AIP) is needed.
In case of a new dependency, check compliance with the ASF 3rd Party License Policy.
In case of backwards incompatible changes please leave a note in a newsfragment file, named {pr_number}.significant.rst or {issue_number}.significant.rst, in newsfragments.

[Lee-W]

The UI looks great! I tested it with the following dag but couldn't see the graph. Is there anything I missed? Thanks!

from __future__ import annotations

import pendulum

from airflow import DAG
from airflow.datasets import Dataset
from airflow.decorators import task

with DAG(
    dag_id="issue_856",
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    schedule=Dataset("s3://bucket/my-task") | Dataset("2") & Dataset("3") | (Dataset("4") & Dataset("5")),
    catchup=False,
    tags=["producer", "dataset"],
):

    @task
    def produce_dataset_events():
        pass

    produce_dataset_events()

[eladkal]

Nice!

[bbovenzi]

The UI looks great! I tested it with the following dag but couldn't see the graph. Is there anything I missed? Thanks!

This is on the DAG graph not the datasets dependency graph. Can you share a screenshot?

[Lee-W]

Yep, it looks like this.

with DAG(
    dag_id="issue_consumer",
    start_date=pendulum.datetime(2021, 1, 1, tz="UTC"),
    schedule=Dataset("1") | Dataset("2"),
    catchup=False,
    tags=["consumer", "dataset"],
):
    ...

[Lee-W]

oh, I finally got what you mean!
It looks super cool. Thanks @bbovenzi !