uranusjr commented on code in PR #49204:
URL: https://github.com/apache/airflow/pull/49204#discussion_r2043449993
##########
airflow-core/src/airflow/models/serialized_dag.py:
##########
@@ -207,26 +207,30 @@ def resolve_asset_uri_ref_dag_dep(self, dep_data: dict)
-> Iterator[DagDependenc
def resolve_asset_alias_dag_dep(self, dep_data: dict) ->
Iterator[DagDependency]:
dep_id = dep_data["dependency_id"]
- for asset_id, asset_name in
self.alias_names_to_asset_ids_names[dep_id]:
- is_source_alias = dep_data["source"] == "asset-alias"
- yield from [
- # asset
- DagDependency(
- source="asset" if is_source_alias else
f"asset-alias:{dep_id}",
- target=f"asset-alias:{dep_id}" if is_source_alias else
"asset",
- label=asset_name,
- dependency_type="asset",
- dependency_id=str(asset_id),
- ),
- # asset alias
- DagDependency(
- source=f"asset:{asset_id}" if is_source_alias else
dep_data["source"],
- target=dep_data["target"] if is_source_alias else
f"asset:{asset_id}",
- label=dep_id,
- dependency_type="asset-alias",
- dependency_id=dep_id,
- ),
- ]
+ assets = self.alias_names_to_asset_ids_names[dep_id]
+ if assets:
+ for asset_id, asset_name in assets:
+ is_source_alias = dep_data["source"] == "asset-alias"
+ yield from [
+ # asset
+ DagDependency(
+ source="asset" if is_source_alias else
f"asset-alias:{dep_id}",
+ target=f"asset-alias:{dep_id}" if is_source_alias else
"asset",
+ label=asset_name,
+ dependency_type="asset",
+ dependency_id=str(asset_id),
+ ),
+ # asset alias
+ DagDependency(
+ source=f"asset:{asset_id}" if is_source_alias else
dep_data["source"],
+ target=dep_data["target"] if is_source_alias else
f"asset:{asset_id}",
+ label=dep_id,
+ dependency_type="asset-alias",
+ dependency_id=dep_id,
+ ),
+ ]
+ else:
+ yield DagDependency(**dep_data)
Review Comment:
> map dataset to asset for 2.x?
This kind of things should probably be handled in a similar way to #49020.
> This `dep_data` was from `DagDependency` before it was serialized. It's
most likely to work fine, but I expanded it as well.
The (reversed) call stack is
* `resolve_asset_alias_dag_dep(dep_data)`
`_DagDependenciesResolver.resolve()`: `dep_data` is a part of
`self.dag_id_dependencies`
* `_DagDependenciesResolver.__init__()`: sets `self.dag_id_dependencies`
* `SerializedDagModel.get_dag_dependencies()`: creates
`_DagDependenciesResolver` with `iterator` loaded from the table
(`serialized_dag`)
So as far as I can tell, this is data from the database after a dag is
serialised.
--
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: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
