You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by "mobuchowski (via GitHub)" <gi...@apache.org> on 2024/10/28 10:41:34 UTC
[PR] serialize asset/dataset timetable conditions in OpenLineage info also for older supported Airflow 2 versions [airflow]
mobuchowski opened a new pull request, #43434:
URL: https://github.com/apache/airflow/pull/43434
Currently, that information is included in OpenLineage events only for Airflow 2.10 and Airflow 3.
--
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] serialize asset/dataset timetable conditions in OpenLineage info also for older supported Airflow 2 versions [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #43434:
URL: https://github.com/apache/airflow/pull/43434#discussion_r1823823541
##########
providers/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -262,9 +262,31 @@ class DagInfo(InfoJsonEncodable):
"start_date",
"tags",
]
- casts = {"timetable": lambda dag: dag.timetable.serialize() if getattr(dag, "timetable", None) else None}
+ casts = {"timetable": lambda dag: DagInfo.serialize_timetable(dag)}
renames = {"_dag_id": "dag_id"}
+ @classmethod
+ def serialize_timetable(cls, dag):
+ serialized = dag.timetable.serialize()
+ if serialized != {} and serialized is not None:
+ return serialized
+ if (
+ hasattr(dag, "dataset_triggers")
+ and isinstance(dag.dataset_triggers, list)
+ and len(dag.dataset_triggers) > 0
+ ):
+ triggers = dag.dataset_triggers
+ return {
+ "dataset_condition": {
Review Comment:
or is it used only in 2.8 and 2.9
--
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] serialize asset/dataset timetable conditions in OpenLineage info also for older supported Airflow 2 versions [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski merged PR #43434:
URL: https://github.com/apache/airflow/pull/43434
--
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] serialize asset/dataset timetable conditions in OpenLineage info also for older supported Airflow 2 versions [airflow]
Posted by "Lee-W (via GitHub)" <gi...@apache.org>.
Lee-W commented on code in PR #43434:
URL: https://github.com/apache/airflow/pull/43434#discussion_r1823823268
##########
providers/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -262,9 +262,31 @@ class DagInfo(InfoJsonEncodable):
"start_date",
"tags",
]
- casts = {"timetable": lambda dag: dag.timetable.serialize() if getattr(dag, "timetable", None) else None}
+ casts = {"timetable": lambda dag: DagInfo.serialize_timetable(dag)}
renames = {"_dag_id": "dag_id"}
+ @classmethod
+ def serialize_timetable(cls, dag):
+ serialized = dag.timetable.serialize()
+ if serialized != {} and serialized is not None:
+ return serialized
+ if (
+ hasattr(dag, "dataset_triggers")
+ and isinstance(dag.dataset_triggers, list)
+ and len(dag.dataset_triggers) > 0
+ ):
+ triggers = dag.dataset_triggers
+ return {
+ "dataset_condition": {
Review Comment:
I think we probably would need to make it asset compatible?
##########
providers/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -262,9 +262,31 @@ class DagInfo(InfoJsonEncodable):
"start_date",
"tags",
]
- casts = {"timetable": lambda dag: dag.timetable.serialize() if getattr(dag, "timetable", None) else None}
+ casts = {"timetable": lambda dag: DagInfo.serialize_timetable(dag)}
renames = {"_dag_id": "dag_id"}
+ @classmethod
+ def serialize_timetable(cls, dag):
+ serialized = dag.timetable.serialize()
+ if serialized != {} and serialized is not None:
+ return serialized
+ if (
+ hasattr(dag, "dataset_triggers")
+ and isinstance(dag.dataset_triggers, list)
+ and len(dag.dataset_triggers) > 0
Review Comment:
```suggestion
and len(dag.dataset_triggers)
```
##########
providers/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -262,9 +262,31 @@ class DagInfo(InfoJsonEncodable):
"start_date",
"tags",
]
- casts = {"timetable": lambda dag: dag.timetable.serialize() if getattr(dag, "timetable", None) else None}
+ casts = {"timetable": lambda dag: DagInfo.serialize_timetable(dag)}
renames = {"_dag_id": "dag_id"}
+ @classmethod
+ def serialize_timetable(cls, dag):
Review Comment:
```suggestion
def serialize_timetable(cls, dag: DAG) -> dict[str, Any]:
```
--
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] serialize asset/dataset timetable conditions in OpenLineage info also for older supported Airflow 2 versions [airflow]
Posted by "mobuchowski (via GitHub)" <gi...@apache.org>.
mobuchowski commented on code in PR #43434:
URL: https://github.com/apache/airflow/pull/43434#discussion_r1827505378
##########
providers/src/airflow/providers/openlineage/utils/utils.py:
##########
@@ -262,9 +262,31 @@ class DagInfo(InfoJsonEncodable):
"start_date",
"tags",
]
- casts = {"timetable": lambda dag: dag.timetable.serialize() if getattr(dag, "timetable", None) else None}
+ casts = {"timetable": lambda dag: DagInfo.serialize_timetable(dag)}
renames = {"_dag_id": "dag_id"}
+ @classmethod
+ def serialize_timetable(cls, dag):
+ serialized = dag.timetable.serialize()
+ if serialized != {} and serialized is not None:
+ return serialized
+ if (
+ hasattr(dag, "dataset_triggers")
+ and isinstance(dag.dataset_triggers, list)
+ and len(dag.dataset_triggers) > 0
+ ):
+ triggers = dag.dataset_triggers
+ return {
+ "dataset_condition": {
Review Comment:
@Lee-W Yes, this is only for 2.8 and 2.9.
--
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