Lee-W commented on code in PR #49204:
URL: https://github.com/apache/airflow/pull/49204#discussion_r2042349283
##########
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:
> Are we sure this is always going to work? I don’t see we blindly
extracting dep_data elsewhere in the code base.
This `dep_data` was from `DagDependency` before it was serialized. It's most
likely to work fine, but I expanded it as well.
> Remember that serialized dags are frozen on serialization, and do not
change with Airflow upgrades.
I thought we were going to reprase the dag?
--
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]
