You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "vincbeck (via GitHub)" <gi...@apache.org> on 2024/11/08 14:54:36 UTC
[PR] AIP-82 Save references between assets and triggers [airflow]
vincbeck opened a new pull request, #43826:
URL: https://github.com/apache/airflow/pull/43826
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`.
<!--
Licensed to the Apache Software Foundation (ASF) under one
or more contributor license agreements. See the NOTICE file
distributed with this work for additional information
regarding copyright ownership. The ASF licenses this file
to you under the Apache License, Version 2.0 (the
"License"); you may not use this file except in compliance
with the License. You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing,
software distributed under the License is distributed on an
"AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
KIND, either express or implied. See the License for the
specific language governing permissions and limitations
under the License.
-->
<!--
Thank you for contributing! Please make sure that your code changes
are covered with tests. And in case of new features or big changes
remember to adjust the documentation.
Feel free to ping committers for the review!
In case of an existing issue, reference it using one of the following:
closes: #ISSUE
related: #ISSUE
How to write a good git commit message:
http://chris.beams.io/posts/git-commit/
-->
<!-- Please keep an empty line above the dashes. -->
---
**^ Add meaningful description above**
Read the **[Pull Request Guidelines](https://github.com/apache/airflow/blob/main/contributing-docs/05_pull_requests.rst#pull-request-guidelines)** for more information.
In case of fundamental code changes, an Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvement+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
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](https://github.com/apache/airflow/tree/main/newsfragments).
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2475109657
> @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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1841804525
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
+
+ asset_model.triggers.append(trigger_model)
+
+ # Remove references from assets no longer used
+ all_assets = session.scalars(select(AssetModel))
+ # orphan_assets = set()
+ for asset_model in all_assets:
+ if (asset_model.name, asset_model.uri) not in self.assets:
+ asset_model.triggers = []
+ # orphan_assets.add(asset_model.id)
Review Comment:
Do we need to do this actively? What happens if we just leave those associations there?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2498369542
> Is there a plan to add tests separate ?
Definitely! It is covered in #42515. However, I did not add unit test for my changes in `collection.py` but I think I should. Though, I'd like to do it in a separate PR. This PR is blocking other in flight PRs I have on that project so unless someone has concerns/feedbacks, I'd like to merge it :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1842600808
##########
airflow/assets/__init__.py:
##########
@@ -266,20 +269,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger] = []
Review Comment:
Makes sense, removed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2474441954
@Lee-W Any chance you can review it? You have some experience around assets that could be interesting to have :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1841798814
##########
airflow/dag_processing/collection.py:
##########
@@ -36,6 +36,7 @@
from airflow.assets import Asset, AssetAlias
from airflow.assets.manager import asset_manager
+from airflow.models import Trigger
Review Comment:
In core Airflow I prefer to always import from the actual definition location instead, i.e. `from airflow.models.trigger import Trigger`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1843310215
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,75 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, dags: dict[str, DagModel], assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ refs_to_add: dict[tuple[str, str], set[str]] = {}
+ refs_to_remove: dict[tuple[str, str], set[str]] = {}
+ triggers: dict[str, BaseTrigger] = {}
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_trigger_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+ triggers.update(trigger_class_path_to_trigger_dict)
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_trigger_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ diff_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ diff_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+ if diff_to_add:
+ refs_to_add[name_uri] = diff_to_add
+ if diff_to_remove:
+ refs_to_remove[name_uri] = diff_to_remove
+
+ if refs_to_add:
+ all_classpaths = {classpath for classpaths in refs_to_add.values() for classpath in classpaths}
+ orm_triggers: dict[str, Trigger] = {
+ trigger.classpath: trigger
+ for trigger in session.scalars(select(Trigger).where(Trigger.classpath.in_(all_classpaths)))
+ }
+
+ # Create new triggers
+ new_trigger_models = [
+ trigger
+ for trigger in [
+ Trigger.from_object(triggers[classpath])
+ for classpath in all_classpaths
+ if classpath not in orm_triggers
+ ]
+ ]
+ session.add_all(new_trigger_models)
+ orm_triggers.update((trigger.classpath, trigger) for trigger in new_trigger_models)
+
+ # Add new references
+ for name_uri, classpaths in refs_to_add.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers.extend(
+ [orm_triggers.get(trigger_class_path) for trigger_class_path in classpaths]
+ )
+
+ if refs_to_remove:
+ # Remove old references
+ for name_uri, classpaths in refs_to_remove.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in classpaths
+ ]
+
+ # Remove references from assets no longer used
+ orphan_assets = session.scalars(
Review Comment:
@uranusjr do we need to check AssetActive here?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1854146504
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,97 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ refs_to_add: dict[tuple[str, str], set[str]] = {}
+ refs_to_remove: dict[tuple[str, str], set[str]] = {}
+ triggers: dict[str, BaseTrigger] = {}
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_repr_to_trigger_dict: dict[str, BaseTrigger] = {
+ repr(trigger): trigger for trigger in asset.watchers
+ }
+ triggers.update(trigger_repr_to_trigger_dict)
+ trigger_repr_from_asset: set[str] = set(trigger_repr_to_trigger_dict.keys())
+
+ trigger_repr_from_asset_model: set[str] = {
+ BaseTrigger.repr(trigger.classpath, trigger.kwargs) for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_repr_from_asset == trigger_repr_from_asset_model:
+ continue
+
+ diff_to_add = trigger_repr_from_asset - trigger_repr_from_asset_model
+ diff_to_remove = trigger_repr_from_asset_model - trigger_repr_from_asset
+ if diff_to_add:
+ refs_to_add[name_uri] = diff_to_add
+ if diff_to_remove:
+ refs_to_remove[name_uri] = diff_to_remove
+
+ if refs_to_add:
+ all_trigger_reprs: set[str] = {
+ trigger_repr for trigger_reprs in refs_to_add.values() for trigger_repr in trigger_reprs
+ }
+
+ all_trigger_keys: list[tuple[str, str]] = [
Review Comment:
`all_trigger_keys` do not contain duplicate because, as you said, it is derived from `refs_to_add`. I guess I could make it a set to be more explicit. Good point :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2498418321
> LGTM. I have played around this table updates, and I believe that every DAG collection call verifies the asset relationships and removes any unused ones. Additionally, asset relationships are only valid during the trigger's lifespan.
>
> This is my understand :) ?
This is correct. Just one clarification here. The trigger lifespan here is as long as the trigger is referenced as watcher of one asset. These triggers will basically live way longer than the current triggers today, those which are associated to a task. But as soon as you remove these references, these triggers will be removed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1834534839
##########
task_sdk/src/airflow/sdk/definitions/dag.py:
##########
@@ -557,9 +557,9 @@ def __lt__(self, other):
def __hash__(self):
hash_components: list[Any] = [type(self)]
for c in _DAG_HASH_ATTRS:
- # task_ids returns a list and lists can't be hashed
- if c == "task_ids":
- val = tuple(self.task_dict)
+ # If it is a list, convert to tuple because lists can't be hashed
Review Comment:
@ashb I made this more generic since `watchers` is also a list and I thought we could just say "if it is a list, then let's convert it to a tuple". Let me know if you have any concerns
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1843685246
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,75 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, dags: dict[str, DagModel], assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ refs_to_add: dict[tuple[str, str], set[str]] = {}
+ refs_to_remove: dict[tuple[str, str], set[str]] = {}
+ triggers: dict[str, BaseTrigger] = {}
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_trigger_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+ triggers.update(trigger_class_path_to_trigger_dict)
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_trigger_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ diff_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ diff_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+ if diff_to_add:
+ refs_to_add[name_uri] = diff_to_add
+ if diff_to_remove:
+ refs_to_remove[name_uri] = diff_to_remove
+
+ if refs_to_add:
+ all_classpaths = {classpath for classpaths in refs_to_add.values() for classpath in classpaths}
+ orm_triggers: dict[str, Trigger] = {
+ trigger.classpath: trigger
+ for trigger in session.scalars(select(Trigger).where(Trigger.classpath.in_(all_classpaths)))
+ }
+
+ # Create new triggers
+ new_trigger_models = [
+ trigger
+ for trigger in [
+ Trigger.from_object(triggers[classpath])
+ for classpath in all_classpaths
+ if classpath not in orm_triggers
+ ]
+ ]
+ session.add_all(new_trigger_models)
+ orm_triggers.update((trigger.classpath, trigger) for trigger in new_trigger_models)
+
+ # Add new references
+ for name_uri, classpaths in refs_to_add.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers.extend(
+ [orm_triggers.get(trigger_class_path) for trigger_class_path in classpaths]
+ )
+
+ if refs_to_remove:
+ # Remove old references
+ for name_uri, classpaths in refs_to_remove.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in classpaths
+ ]
+
+ # Remove references from assets no longer used
+ orphan_assets = session.scalars(
Review Comment:
An asset without an AssetActive entry is not referenced anywhere, and trigerring an event to such an asset will therefore simply do nothing. So _not_ checking AssetActive here is not useful in practice, but maybe theoratically a possibility? It depends on what we want the user to be able to do, I guess. @vincbeck Do you think a user should be able to trigger an event on an asset that does not actually exist in any DAGs?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1842835117
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
+
+ asset_model.triggers.append(trigger_model)
+
+ # Remove references from assets no longer used
+ all_assets = session.scalars(select(AssetModel))
+ # orphan_assets = set()
+ for asset_model in all_assets:
+ if (asset_model.name, asset_model.uri) not in self.assets:
+ asset_model.triggers = []
+ # orphan_assets.add(asset_model.id)
Review Comment:
Fixed
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1843629522
##########
airflow/assets/__init__.py:
##########
@@ -276,20 +279,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger]
asset_type: ClassVar[str] = "asset"
__version__: ClassVar[int] = 1
@overload
- def __init__(self, name: str, uri: str, *, group: str = "", extra: dict | None = None) -> None:
+ def __init__(
+ self,
+ name: str,
+ uri: str,
+ *,
+ group: str = "",
+ extra: dict | None = None,
+ watchers: list[BaseTrigger] | None = None,
Review Comment:
I wonder if *watcher* is a good name for this. What do we expect this to do? If I understand AIP-82 correctly, an external event would fire the trigger, and the trigger would create events for assets associated to it.
Assuming my understanding is correct, the triggers here are not watchers of the asset; rather, the asset *watches* the triggers. The relationship is the other way around. So it is probably better to call this `watch` instead? Or maybe this attribute should live on the trigger instead, something like
```python
asset = Asset("example_asset_watchers")
trigger = SqsSensorTrigger(sqs_queue="my_queue", trigger=[asset])
DAG(..., schedule=[asset])
```
Tell me what you think on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1842782395
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
Review Comment:
Thank you for your suggestions, I tried to apply them. Please let me know if this is what you thought
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1851771313
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,75 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, dags: dict[str, DagModel], assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ refs_to_add: dict[tuple[str, str], set[str]] = {}
+ refs_to_remove: dict[tuple[str, str], set[str]] = {}
+ triggers: dict[str, BaseTrigger] = {}
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_trigger_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+ triggers.update(trigger_class_path_to_trigger_dict)
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_trigger_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ diff_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ diff_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+ if diff_to_add:
+ refs_to_add[name_uri] = diff_to_add
+ if diff_to_remove:
+ refs_to_remove[name_uri] = diff_to_remove
+
+ if refs_to_add:
+ all_classpaths = {classpath for classpaths in refs_to_add.values() for classpath in classpaths}
+ orm_triggers: dict[str, Trigger] = {
+ trigger.classpath: trigger
+ for trigger in session.scalars(select(Trigger).where(Trigger.classpath.in_(all_classpaths)))
+ }
+
+ # Create new triggers
+ new_trigger_models = [
+ trigger
+ for trigger in [
+ Trigger.from_object(triggers[classpath])
+ for classpath in all_classpaths
+ if classpath not in orm_triggers
+ ]
+ ]
+ session.add_all(new_trigger_models)
+ orm_triggers.update((trigger.classpath, trigger) for trigger in new_trigger_models)
+
+ # Add new references
+ for name_uri, classpaths in refs_to_add.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers.extend(
+ [orm_triggers.get(trigger_class_path) for trigger_class_path in classpaths]
+ )
+
+ if refs_to_remove:
+ # Remove old references
+ for name_uri, classpaths in refs_to_remove.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in classpaths
+ ]
+
+ # Remove references from assets no longer used
+ orphan_assets = session.scalars(
Review Comment:
> I could use AssetActive here but I actually need inactive assets, thus I could do something like "all_assets - active_assets". I am not sure this is better than the way I do it right now
sounds reasonable š¤
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1841802344
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
Review Comment:
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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1847351749
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,75 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, dags: dict[str, DagModel], assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ refs_to_add: dict[tuple[str, str], set[str]] = {}
+ refs_to_remove: dict[tuple[str, str], set[str]] = {}
+ triggers: dict[str, BaseTrigger] = {}
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_trigger_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+ triggers.update(trigger_class_path_to_trigger_dict)
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_trigger_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ diff_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ diff_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+ if diff_to_add:
+ refs_to_add[name_uri] = diff_to_add
+ if diff_to_remove:
+ refs_to_remove[name_uri] = diff_to_remove
+
+ if refs_to_add:
+ all_classpaths = {classpath for classpaths in refs_to_add.values() for classpath in classpaths}
+ orm_triggers: dict[str, Trigger] = {
+ trigger.classpath: trigger
+ for trigger in session.scalars(select(Trigger).where(Trigger.classpath.in_(all_classpaths)))
+ }
+
+ # Create new triggers
+ new_trigger_models = [
+ trigger
+ for trigger in [
+ Trigger.from_object(triggers[classpath])
+ for classpath in all_classpaths
+ if classpath not in orm_triggers
+ ]
+ ]
+ session.add_all(new_trigger_models)
+ orm_triggers.update((trigger.classpath, trigger) for trigger in new_trigger_models)
+
+ # Add new references
+ for name_uri, classpaths in refs_to_add.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers.extend(
+ [orm_triggers.get(trigger_class_path) for trigger_class_path in classpaths]
+ )
+
+ if refs_to_remove:
+ # Remove old references
+ for name_uri, classpaths in refs_to_remove.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in classpaths
+ ]
+
+ # Remove references from assets no longer used
+ orphan_assets = session.scalars(
Review Comment:
I could use `AssetActive` here but I actually need inactive assets, thus I could do something like "all_assets - active_assets". I am not sure this is better than the way I do it right now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1843963894
##########
airflow/assets/__init__.py:
##########
@@ -276,20 +279,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger]
asset_type: ClassVar[str] = "asset"
__version__: ClassVar[int] = 1
@overload
- def __init__(self, name: str, uri: str, *, group: str = "", extra: dict | None = None) -> None:
+ def __init__(
+ self,
+ name: str,
+ uri: str,
+ *,
+ group: str = "",
+ extra: dict | None = None,
+ watchers: list[BaseTrigger] | None = None,
Review Comment:
Naming ... so hard haha. I see your point.
The reason why I called it `watchers` is because the triggers will `watch` some external resource and send event on updates. In that sense, to me, the triggers are watchers. I am not strongly against `watch` if you think it makes more sense. To be very honest, between `watchers` and `watch` I dont mind, I think the both of them makes sense.
However, I definitely want the attribute on the asset class, I think it makes more sense and a more deliberate choice for the user to say, I have this asset and I want this asset to be updated when these triggers fire.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2466046072
> @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.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2470846458
> > @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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1842783220
##########
airflow/dag_processing/collection.py:
##########
@@ -36,6 +36,7 @@
from airflow.assets import Asset, AssetAlias
from airflow.assets.manager import asset_manager
+from airflow.models import Trigger
Review Comment:
Good point, feel like there should be a pre-commit for that
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1843621102
##########
airflow/assets/__init__.py:
##########
@@ -276,20 +279,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger]
asset_type: ClassVar[str] = "asset"
__version__: ClassVar[int] = 1
@overload
- def __init__(self, name: str, uri: str, *, group: str = "", extra: dict | None = None) -> None:
+ def __init__(
+ self,
+ name: str,
+ uri: str,
+ *,
+ group: str = "",
+ extra: dict | None = None,
+ watchers: list[BaseTrigger] | None = None,
Review Comment:
How is this argument supposed to be used? This class is a user-facing interface; do we expect users to pass in triggers manually when they create an asset? If not, we should not add this argument here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1854157839
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,97 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ refs_to_add: dict[tuple[str, str], set[str]] = {}
+ refs_to_remove: dict[tuple[str, str], set[str]] = {}
+ triggers: dict[str, BaseTrigger] = {}
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_repr_to_trigger_dict: dict[str, BaseTrigger] = {
+ repr(trigger): trigger for trigger in asset.watchers
+ }
+ triggers.update(trigger_repr_to_trigger_dict)
+ trigger_repr_from_asset: set[str] = set(trigger_repr_to_trigger_dict.keys())
+
+ trigger_repr_from_asset_model: set[str] = {
+ BaseTrigger.repr(trigger.classpath, trigger.kwargs) for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_repr_from_asset == trigger_repr_from_asset_model:
+ continue
+
+ diff_to_add = trigger_repr_from_asset - trigger_repr_from_asset_model
+ diff_to_remove = trigger_repr_from_asset_model - trigger_repr_from_asset
+ if diff_to_add:
+ refs_to_add[name_uri] = diff_to_add
+ if diff_to_remove:
+ refs_to_remove[name_uri] = diff_to_remove
+
+ if refs_to_add:
+ all_trigger_reprs: set[str] = {
+ trigger_repr for trigger_reprs in refs_to_add.values() for trigger_repr in trigger_reprs
+ }
+
+ all_trigger_keys: list[tuple[str, str]] = [
Review Comment:
I updated it to make it a set
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1846861559
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
Review Comment:
What do you think?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1841794966
##########
airflow/assets/__init__.py:
##########
@@ -266,20 +269,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger] = []
Review Comment:
It does not handle default values well. This should use `attrs.field(factory=list)` instead.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1842792069
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
+
+ asset_model.triggers.append(trigger_model)
+
+ # Remove references from assets no longer used
+ all_assets = session.scalars(select(AssetModel))
+ # orphan_assets = set()
+ for asset_model in all_assets:
+ if (asset_model.name, asset_model.uri) not in self.assets:
+ asset_model.triggers = []
+ # orphan_assets.add(asset_model.id)
Review Comment:
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
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1851772574
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
Review Comment:
look better IMO
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck merged PR #43826:
URL: https://github.com/apache/airflow/pull/43826
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1843953549
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,75 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, dags: dict[str, DagModel], assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ refs_to_add: dict[tuple[str, str], set[str]] = {}
+ refs_to_remove: dict[tuple[str, str], set[str]] = {}
+ triggers: dict[str, BaseTrigger] = {}
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_trigger_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+ triggers.update(trigger_class_path_to_trigger_dict)
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_trigger_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ diff_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ diff_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+ if diff_to_add:
+ refs_to_add[name_uri] = diff_to_add
+ if diff_to_remove:
+ refs_to_remove[name_uri] = diff_to_remove
+
+ if refs_to_add:
+ all_classpaths = {classpath for classpaths in refs_to_add.values() for classpath in classpaths}
+ orm_triggers: dict[str, Trigger] = {
+ trigger.classpath: trigger
+ for trigger in session.scalars(select(Trigger).where(Trigger.classpath.in_(all_classpaths)))
+ }
+
+ # Create new triggers
+ new_trigger_models = [
+ trigger
+ for trigger in [
+ Trigger.from_object(triggers[classpath])
+ for classpath in all_classpaths
+ if classpath not in orm_triggers
+ ]
+ ]
+ session.add_all(new_trigger_models)
+ orm_triggers.update((trigger.classpath, trigger) for trigger in new_trigger_models)
+
+ # Add new references
+ for name_uri, classpaths in refs_to_add.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers.extend(
+ [orm_triggers.get(trigger_class_path) for trigger_class_path in classpaths]
+ )
+
+ if refs_to_remove:
+ # Remove old references
+ for name_uri, classpaths in refs_to_remove.items():
+ asset_model = assets[name_uri]
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in classpaths
+ ]
+
+ # Remove references from assets no longer used
+ orphan_assets = session.scalars(
Review Comment:
Interesting, I did not know that notion of `AssetActive`, maybe I could use it.
> Do you think a user should be able to trigger an event on an asset that does not actually exist in any DAGs?
Absolutely not, that's what I am doing (or trying to do) here but even further. `orphan_assets` contains all the assets not used by any DAGs as schedule. In other words, no DAG use an asset in `orphan_assets` as schedule condition. I am removing all references from these assets since they are not used to schedule DAG. The way I understand it is, all assets with an `AssetActive` entry right is a subset of `orphan_assets` ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2498622048
> > LGTM. I have played around this table updates, and I believe that every DAG collection call verifies the asset relationships and removes any unused ones. Additionally, asset relationships are only valid during the trigger's lifespan.
> > This is my understand :) ?
>
> This is correct. Just one clarification here. The trigger lifespan here is as long as the trigger is referenced as watcher of one asset. These triggers will basically live way longer than the current triggers today, those which are associated to a task. But as soon as you remove these references, these triggers will be removed
cool, thank you :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1856352710
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,97 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ refs_to_add: dict[tuple[str, str], set[str]] = {}
+ refs_to_remove: dict[tuple[str, str], set[str]] = {}
+ triggers: dict[str, BaseTrigger] = {}
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_repr_to_trigger_dict: dict[str, BaseTrigger] = {
+ repr(trigger): trigger for trigger in asset.watchers
+ }
+ triggers.update(trigger_repr_to_trigger_dict)
+ trigger_repr_from_asset: set[str] = set(trigger_repr_to_trigger_dict.keys())
+
+ trigger_repr_from_asset_model: set[str] = {
+ BaseTrigger.repr(trigger.classpath, trigger.kwargs) for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_repr_from_asset == trigger_repr_from_asset_model:
+ continue
+
+ diff_to_add = trigger_repr_from_asset - trigger_repr_from_asset_model
+ diff_to_remove = trigger_repr_from_asset_model - trigger_repr_from_asset
+ if diff_to_add:
+ refs_to_add[name_uri] = diff_to_add
+ if diff_to_remove:
+ refs_to_remove[name_uri] = diff_to_remove
+
+ if refs_to_add:
+ all_trigger_reprs: set[str] = {
+ trigger_repr for trigger_reprs in refs_to_add.values() for trigger_repr in trigger_reprs
+ }
+
+ all_trigger_keys: list[tuple[str, str]] = [
Review Comment:
Thanks LGTM.. :)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2498389691
> > Is there a plan to add tests separate ?
>
> Definitely! It is covered in #42515. However, I did not add unit test for my changes in `collection.py` but I think I should. Though, I'd like to do it in a separate PR. This PR is blocking other in flight PRs I have on that project so unless someone has concerns/feedbacks, I'd like to merge it :)
ah okay make sense i didnt see that task.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1841686146
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
Review Comment:
Not sure whether collect all the models together and use `add_all` would be better š¤
##########
airflow/assets/__init__.py:
##########
@@ -301,10 +328,14 @@ def __init__(
self.uri = _sanitize_uri(_validate_non_empty_identifier(self, fields["uri"], uri))
self.group = _validate_identifier(self, fields["group"], group) if group else self.asset_type
self.extra = _set_extra_default(extra)
+ self.watchers = watchers or []
def __fspath__(self) -> str:
return self.uri
+ def __hash__(self) -> int:
Review Comment:
It was removed after discussing with @uranusjr . seems to introduce more trouble than benefit then. is there any reason we want to add it here?
##########
airflow/assets/__init__.py:
##########
@@ -266,20 +269,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger] = []
Review Comment:
do we need to use field instead? not sure how attrs deals with default values
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
+
+ asset_model.triggers.append(trigger_model)
+
+ # Remove references from assets no longer used
+ all_assets = session.scalars(select(AssetModel))
+ # orphan_assets = set()
+ for asset_model in all_assets:
+ if (asset_model.name, asset_model.uri) not in self.assets:
+ asset_model.triggers = []
+ # orphan_assets.add(asset_model.id)
Review Comment:
```suggestion
```
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1841796789
##########
airflow/assets/__init__.py:
##########
@@ -266,20 +269,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger] = []
Review Comment:
Although I do not think we need a default at all? The `__init__` implementation always provides a value anyway.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1852893114
##########
airflow/assets/__init__.py:
##########
@@ -276,20 +279,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger]
asset_type: ClassVar[str] = "asset"
__version__: ClassVar[int] = 1
@overload
- def __init__(self, name: str, uri: str, *, group: str = "", extra: dict | None = None) -> None:
+ def __init__(
+ self,
+ name: str,
+ uri: str,
+ *,
+ group: str = "",
+ extra: dict | None = None,
+ watchers: list[BaseTrigger] | None = None,
Review Comment:
What do you think @uranusjr ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2465621190
@Lee-W @uranusjr When working on it realized that assets are added in the DB from DAG definition when 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?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1842617394
##########
airflow/assets/__init__.py:
##########
@@ -301,10 +328,14 @@ def __init__(
self.uri = _sanitize_uri(_validate_non_empty_identifier(self, fields["uri"], uri))
self.group = _validate_identifier(self, fields["group"], group) if group else self.asset_type
self.extra = _set_extra_default(extra)
+ self.watchers = watchers or []
def __fspath__(self) -> str:
return self.uri
+ def __hash__(self) -> int:
Review Comment:
I am sure I needed it when implementing it but I just tested it without it and it worked so I'll remove it
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1842625730
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,51 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_class_path_to_asset_dict: dict[str, BaseTrigger] = {
+ trigger.serialize()[0]: trigger for trigger in asset.watchers
+ }
+
+ trigger_class_paths_from_asset: set[str] = set(trigger_class_path_to_asset_dict.keys())
+ trigger_class_paths_from_asset_model: set[str] = {
+ trigger.classpath for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_class_paths_from_asset == trigger_class_paths_from_asset_model:
+ continue
+
+ refs_to_add = trigger_class_paths_from_asset - trigger_class_paths_from_asset_model
+ refs_to_remove = trigger_class_paths_from_asset_model - trigger_class_paths_from_asset
+
+ # Remove old references
+ asset_model.triggers = [
+ trigger for trigger in asset_model.triggers if trigger.classpath not in refs_to_remove
+ ]
+
+ # Add new references
+ for trigger_class_path in refs_to_add:
+ trigger_model = session.scalar(
+ select(Trigger).where(Trigger.classpath == trigger_class_path).limit(1)
+ )
+
+ # Create the trigger in the DB if it does not exist
+ if not trigger_model:
+ trigger_model = Trigger.from_object(trigger_class_path_to_asset_dict[trigger_class_path])
+ session.add(trigger_model)
+
+ asset_model.triggers.append(trigger_model)
+
+ # Remove references from assets no longer used
+ all_assets = session.scalars(select(AssetModel))
+ # orphan_assets = set()
+ for asset_model in all_assets:
+ if (asset_model.name, asset_model.uri) not in self.assets:
+ asset_model.triggers = []
+ # orphan_assets.add(asset_model.id)
Review Comment:
Thanks :D
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1854088949
##########
airflow/dag_processing/collection.py:
##########
@@ -425,3 +427,97 @@ def add_task_asset_references(
for task_id, asset_id in referenced_outlets
if (task_id, asset_id) not in orm_refs
)
+
+ def add_asset_trigger_references(
+ self, assets: dict[tuple[str, str], AssetModel], *, session: Session
+ ) -> None:
+ # Update references from assets being used
+ refs_to_add: dict[tuple[str, str], set[str]] = {}
+ refs_to_remove: dict[tuple[str, str], set[str]] = {}
+ triggers: dict[str, BaseTrigger] = {}
+ for name_uri, asset in self.assets.items():
+ asset_model = assets[name_uri]
+ trigger_repr_to_trigger_dict: dict[str, BaseTrigger] = {
+ repr(trigger): trigger for trigger in asset.watchers
+ }
+ triggers.update(trigger_repr_to_trigger_dict)
+ trigger_repr_from_asset: set[str] = set(trigger_repr_to_trigger_dict.keys())
+
+ trigger_repr_from_asset_model: set[str] = {
+ BaseTrigger.repr(trigger.classpath, trigger.kwargs) for trigger in asset_model.triggers
+ }
+
+ # Optimization: no diff between the DB and DAG definitions, no update needed
+ if trigger_repr_from_asset == trigger_repr_from_asset_model:
+ continue
+
+ diff_to_add = trigger_repr_from_asset - trigger_repr_from_asset_model
+ diff_to_remove = trigger_repr_from_asset_model - trigger_repr_from_asset
+ if diff_to_add:
+ refs_to_add[name_uri] = diff_to_add
+ if diff_to_remove:
+ refs_to_remove[name_uri] = diff_to_remove
+
+ if refs_to_add:
+ all_trigger_reprs: set[str] = {
+ trigger_repr for trigger_reprs in refs_to_add.values() for trigger_repr in trigger_reprs
+ }
+
+ all_trigger_keys: list[tuple[str, str]] = [
Review Comment:
Is `all_trigger_keys` contain duplicates? If so, are the duplicates necessary, or is there something I might be missing? :)
I noticed that `all_trigger_reprs` contains only unique elements, so Iām wondering if the same should apply to all_trigger_keys, especially since both are derived from the values in `refs_to_add`.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "gopidesupavan (via GitHub)" <gi...@apache.org>.
gopidesupavan commented on PR #43826:
URL: https://github.com/apache/airflow/pull/43826#issuecomment-2497645890
Is there a plan to add tests separate ?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1856891889
##########
airflow/assets/__init__.py:
##########
@@ -276,20 +279,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger]
asset_type: ClassVar[str] = "asset"
__version__: ClassVar[int] = 1
@overload
- def __init__(self, name: str, uri: str, *, group: str = "", extra: dict | None = None) -> None:
+ def __init__(
+ self,
+ name: str,
+ uri: str,
+ *,
+ group: str = "",
+ extra: dict | None = None,
+ watchers: list[BaseTrigger] | None = None,
Review Comment:
I did not hear back so I'll go with this. If any concerns or feedbacks, I can always adjust later
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "uranusjr (via GitHub)" <gi...@apache.org>.
uranusjr commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1843621102
##########
airflow/assets/__init__.py:
##########
@@ -276,20 +279,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger]
asset_type: ClassVar[str] = "asset"
__version__: ClassVar[int] = 1
@overload
- def __init__(self, name: str, uri: str, *, group: str = "", extra: dict | None = None) -> None:
+ def __init__(
+ self,
+ name: str,
+ uri: str,
+ *,
+ group: str = "",
+ extra: dict | None = None,
+ watchers: list[BaseTrigger] | None = None,
Review Comment:
How is this argument supposed to be used? This class is a user-facing interface; do we expect users to pass in triggers manually when they create an asset? If not, we should not add this argument here.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
Re: [PR] AIP-82 Save references between assets and triggers [airflow]
Posted by "vincbeck (via GitHub)" <gi...@apache.org>.
vincbeck commented on code in PR #43826:
URL: https://github.com/apache/airflow/pull/43826#discussion_r1843963894
##########
airflow/assets/__init__.py:
##########
@@ -276,20 +279,43 @@ class Asset(os.PathLike, BaseAsset):
uri: str
group: str
extra: dict[str, Any]
+ watchers: list[BaseTrigger]
asset_type: ClassVar[str] = "asset"
__version__: ClassVar[int] = 1
@overload
- def __init__(self, name: str, uri: str, *, group: str = "", extra: dict | None = None) -> None:
+ def __init__(
+ self,
+ name: str,
+ uri: str,
+ *,
+ group: str = "",
+ extra: dict | None = None,
+ watchers: list[BaseTrigger] | None = None,
Review Comment:
Naming ... so hard haha. I see your point.
The reason why I called it `watchers` is because the triggers will `watch` some external resource and send event on updates. In that sense, to me, the triggers are watchers. I am not strongly again `watch` if you think it makes more sense. To be very honest, between `watchers` and `watch` I dont mind, I think the both of them makes sense.
However, I definitely want the attribute on the asset class, I think it makes more sense and a more deliberate choice for the user to say, I have this asset and I want this asset to be updated when these triggers fire.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: commits-unsubscribe@airflow.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org