AIP-82 Save references between assets and triggers

[vincbeck]

Resolves #42510.

This PR adds a new attributes watchers to the Asset class and saves references between assets and triggers in the DB. For example:

trigger = SqsSensorTrigger(sqs_queue="my_queue")
asset = Asset("example_asset_watchers", watchers=[trigger])

with DAG(
    dag_id="example_dataset_watcher",
    schedule=[asset],
    catchup=False,
):
    task = EmptyOperator(task_id="task",)

    chain(task)

This PR creates the trigger in the DB if it does not exist and save the reference between asset and trigger.

---

^ 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.

[vincbeck]

@Lee-W @uranusjr When working on it I realized that assets are added in the DB from DAG definition but never removed (or at least I did not see the code). Meaning, as a DAG author if I define an asset in my DAG and then later on remove it, the asset is never removed from the DB. Am I wrong? If not, is it intended?

[Lee-W]

@Lee-W @uranusjr When working on it I realized that assets are added in the DB from DAG definition but never removed (or at least I did not see the code). Meaning, as a DAG author if I define an asset in my DAG and then later on remove it, the asset is never removed from the DB. Am I wrong? If not, is it intended?

Yep, this is by design as of now. To keep the asset history.

[vincbeck]

@Lee-W @uranusjr When working on it I realized that assets are added in the DB from DAG definition but never removed (or at least I did not see the code). Meaning, as a DAG author if I define an asset in my DAG and then later on remove it, the asset is never removed from the DB. Am I wrong? If not, is it intended?

Yep, this is by design as of now. To keep the asset history.

Alright, thank you. I handled it then. I removed the references from asset and triggers if the asset is no longer used

[vincbeck]

@Lee-W Any chance you can review it? You have some experience around assets that could be interesting to have :)

[Lee-W]

@Lee-W Any chance you can review it? You have some experience around assets that could be interesting to have :)

Sure thing :) Will take a look later today

[Lee-W]

Not sure whether collect all the models together and use add_all would be better 🤔

[uranusjr]

Collecting all model objects first is cleaner code IMO; this loop + add + append approach is a lot more difficult to read. Also the repeated scalar + limit call is not very performant; it is better to select all the existing triggers first in one query.

[vincbeck]

Thank you for your suggestions, I tried to apply them. Please let me know if this is what you thought

[uranusjr]

Do we need to do this actively? What happens if we just leave those associations there?

[vincbeck]

Then the trigger will keep updating the asset in cases of events. More importantly, if we keep the association between the asset and the trigger, it will be impossible to clean-up these triggers. I want to be able to remove triggers that are not used (meaning, not associated to a task and an asset). Which means they will keep infinitely pooling an external resource. That could be very costly.

On that same topic, when doing some testing, I noticed that this function is called per DAG (am I wrong?). As a consequence, this piece of code removes the associations I just created before. I need to fix that

[vincbeck]

Fixed