uranusjr commented on code in PR #37826:
URL: https://github.com/apache/airflow/pull/37826#discussion_r1509917812
##########
airflow/api_connexion/schemas/dag_schema.py:
##########
@@ -148,6 +150,37 @@ def get_params(obj: DAG):
params = obj.params
return {k: v.dump() for k, v in params.items()}
+ @staticmethod
+ def get_dataset_expression(obj: DAG):
+ """Convert the dataset_triggers structure from SerializedDagModel into
a dataset_expression."""
+ serialized_dag_model = SerializedDagModel.get(obj.dag_id)
+ if serialized_dag_model:
+ dag_data = serialized_dag_model.data or {}
+ dataset_triggers = dag_data.get("dag", {}).get("dataset_triggers")
+ if isinstance(dataset_triggers, dict):
+ return DAGDetailSchema._parse_dataset_trigger(dataset_triggers)
+ return {}
+
+ @staticmethod
+ def _parse_dataset_trigger(trigger: dict) -> str | dict:
+ """
+ Recursively parse the dataset trigger to build the dataset expression.
+
+ :param trigger: The dataset_triggers to parse
+ """
+ # Direct dataset reference
+ if trigger.get("__type") == "dataset":
+ return trigger["__var"]["uri"]
+
+ # Compound dataset ('any' or 'all')
+ elif "dataset_" in trigger.get("__type", ""): # Checks if type
contains 'dataset_'
+ # Extract 'any' or 'all' from '__type'
+ expr_type = trigger["__type"].split("_")[1]
+ return {expr_type: [DAGDetailSchema._parse_dataset_trigger(t) for
t in trigger["__var"]]}
+
+ # If the type is neither 'dataset' nor starts with 'dataset_', raise
an error
+ raise ValueError(f"Unsupported dataset trigger type:
{trigger.get('__type')}")
Review Comment:
This feels like just reimplementing `BaseSerialization.deserialize`. Can we
not just use that or something else in `serialized_objects`?
--
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]
