This is an automated email from the ASF dual-hosted git repository.
mobuchowski pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/airflow.git
The following commit(s) were added to refs/heads/main by this push:
new 6f358052a42 Send task inlets and outlets as JSON lists in OpenLineage
facet (#74666)
6f358052a42 is described below
commit 6f358052a423abfd9eacc128680089cef1885c0d
Author: Frank Ye <[email protected]>
AuthorDate: Mon Oct 19 05:23:24 2026 -0600
Send task inlets and outlets as JSON lists in OpenLineage facet (#74666)
* Send task inlets and outlets as JSON lists in OpenLineage facet
* Add list keys to OpenLineage system test events and keep outlets with bad
keys
* Make OpenLineage asset list tests pass on Airflow 2 and 3.0
---
.../airflow/providers/openlineage/utils/utils.py | 19 ++++
.../openlineage_base_complex_dag__af2.json | 56 +++++++++++
.../openlineage_base_complex_dag__af3.json | 62 +++++++++++++
.../openlineage_base_simple_dag__af2.json | 4 +
.../openlineage_base_simple_dag__af3.json | 4 +
.../openlineage_custom_operator_failure_dag.json | 2 +
.../openlineage_manual_lineage_dag.json | 4 +
.../openlineage_taskflow_simple_dag.json | 4 +
.../tests/unit/openlineage/utils/test_utils.py | 102 ++++++++++++++++++++-
9 files changed, 256 insertions(+), 1 deletion(-)
diff --git
a/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
b/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
index 6c53cc6dcfe..ca43441870d 100644
--- a/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
+++ b/providers/openlineage/src/airflow/providers/openlineage/utils/utils.py
@@ -1298,6 +1298,12 @@ class TaskInfo(InfoJsonEncodable):
def __init__(self, obj, runtime_assets: list[tuple[Asset, AssetAlias |
None]] | None = None):
self._runtime_assets = runtime_assets # Will not be included in final
dict, not in self._fields.
super().__init__(obj)
+ # ``inlets`` / ``outlets`` are a Python repr string that consumers
cannot reliably parse
+ # (``extra`` can hold quotes, ``Decimal``, ``True``, ...). Expose them
as real lists too.
+ # Set after ``super().__init__`` so ``_cast_basic_types`` does not
turn them into strings.
+ for field in ("inlets", "outlets"):
+ if hasattr(self, field):
+ self[f"{field}_list"] =
_convert_to_json_safe_list(getattr(self, field), field)
def _extend_fields(self) -> None:
if not self._runtime_assets:
@@ -2203,6 +2209,19 @@ class OpenLineageRedactor(SecretsMasker):
return item
+def _convert_to_json_safe_list(value: list, name: str) -> list | None:
+ """
+ Return ``value`` with only JSON-native types.
+
+ Non-serializable values are cast with ``str()``; dict keys JSON can't
encode (e.g. tuples) are dropped.
+ """
+ try:
+ return json.loads(json.dumps(value, default=str, skipkeys=True))
+ except (TypeError, ValueError) as e:
+ log.warning("Unable to serialize task `%s` to a JSON list: %s", name,
e)
+ return None
+
+
def is_json_serializable(item):
try:
json.dumps(item)
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af2.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af2.json
index 413af8d8935..4e22417b9a4 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af2.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af2.json
@@ -331,6 +331,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -339,6 +340,7 @@
"operator_class_path":
"airflow.providers.standard.operators.bash.BashOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "owner'2",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -509,6 +511,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -517,6 +520,7 @@
"operator_class_path":
"airflow.providers.standard.operators.bash.BashOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "owner'2",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -685,6 +689,16 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[{'uri': 'file://host1/dir2/file2.txt',
'extra': None, 'type': 'asset'}, {'uri':
'unknown_scheme://host1/dir2/file3.txt', 'extra': None, 'type': 'asset'}]",
+ "inlets_list": [
+ {
+ "uri": "file://host1/dir2/file2.txt",
+ "type": "asset"
+ },
+ {
+ "uri": "unknown_scheme://host1/dir2/file3.txt",
+ "type": "asset"
+ }
+ ],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -694,6 +708,7 @@
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -867,6 +882,16 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[{'uri': 'file://host1/dir2/file2.txt',
'extra': None, 'type': 'asset'}, {'uri':
'unknown_scheme://host1/dir2/file3.txt', 'extra': None, 'type': 'asset'}]",
+ "inlets_list": [
+ {
+ "uri": "file://host1/dir2/file2.txt",
+ "type": "asset"
+ },
+ {
+ "uri": "unknown_scheme://host1/dir2/file3.txt",
+ "type": "asset"
+ }
+ ],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -876,6 +901,7 @@
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1042,6 +1068,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -1050,6 +1077,12 @@
"operator_class_path":
"airflow.providers.standard.operators.empty.EmptyOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[{'uri': 'file://host/dir/file.txt',
'extra': None, 'type': 'asset'}]",
+ "outlets_list": [
+ {
+ "uri": "file://host/dir/file.txt",
+ "type": "asset"
+ }
+ ],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1217,6 +1250,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -1225,6 +1259,12 @@
"operator_class_path":
"airflow.providers.standard.operators.empty.EmptyOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[{'uri': 'file://host/dir/file.txt',
'extra': None, 'type': 'asset'}]",
+ "outlets_list": [
+ {
+ "uri": "file://host/dir/file.txt",
+ "type": "asset"
+ }
+ ],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1388,6 +1428,7 @@
"external_task_id": "external_task_id",
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -1397,6 +1438,7 @@
"operator_class": "SomeCustomOperator",
"operator_class_path": "{{
result.endswith('openlineage_base_complex_dag.SomeCustomOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "owner3",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1552,6 +1594,7 @@
"external_task_id": "external_task_id",
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -1561,6 +1604,7 @@
"operator_class": "SomeCustomOperator",
"operator_class_path": "{{
result.endswith('openlineage_base_complex_dag.SomeCustomOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "owner3",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1712,6 +1756,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": "{{ result == true }}",
@@ -1719,6 +1764,7 @@
"operator_class": "CustomMappedOperator",
"operator_class_path": "{{
result.endswith('openlineage_base_complex_dag.CustomMappedOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1880,6 +1926,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": "{{ result == true }}",
@@ -1887,6 +1934,7 @@
"operator_class": "CustomMappedOperator",
"operator_class_path": "{{
result.endswith('openlineage_base_complex_dag.CustomMappedOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -2047,6 +2095,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -2055,6 +2104,7 @@
"operator_class_path":
"airflow.providers.standard.operators.empty.EmptyOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"sla": "123.0 seconds",
@@ -2226,6 +2276,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -2234,6 +2285,7 @@
"operator_class_path":
"airflow.providers.standard.operators.empty.EmptyOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"sla": "123.0 seconds",
@@ -2404,6 +2456,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -2412,6 +2465,7 @@
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -2581,6 +2635,7 @@
"executor_config": {},
"ignore_first_depends_on_past": "{{ result == true }}",
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -2589,6 +2644,7 @@
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af3.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af3.json
index 1c0d509d3ba..e7527369849 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af3.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_complex_dag__af3.json
@@ -332,12 +332,14 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "BashOperator",
"operator_class_path":
"airflow.providers.standard.operators.bash.BashOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "owner'2",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -505,12 +507,14 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "BashOperator",
"operator_class_path":
"airflow.providers.standard.operators.bash.BashOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "owner'2",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -677,6 +681,18 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[{'uri': 'file://host1/dir2/file2.txt',
'extra': {}, 'type': 'asset'}, {'uri': 'unknown_scheme://host1/dir2/file3.txt',
'extra': {}, 'type': 'asset'}]",
+ "inlets_list": [
+ {
+ "uri": "file://host1/dir2/file2.txt",
+ "extra": {},
+ "type": "asset"
+ },
+ {
+ "uri": "unknown_scheme://host1/dir2/file3.txt",
+ "extra": {},
+ "type": "asset"
+ }
+ ],
"mapped": false,
"max_retry_delay": "42.0 seconds",
"multiple_outputs": false,
@@ -684,6 +700,7 @@
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -854,6 +871,18 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[{'uri': 'file://host1/dir2/file2.txt',
'extra': {}, 'type': 'asset'}, {'uri': 'unknown_scheme://host1/dir2/file3.txt',
'extra': {}, 'type': 'asset'}]",
+ "inlets_list": [
+ {
+ "uri": "file://host1/dir2/file2.txt",
+ "extra": {},
+ "type": "asset"
+ },
+ {
+ "uri": "unknown_scheme://host1/dir2/file3.txt",
+ "extra": {},
+ "type": "asset"
+ }
+ ],
"mapped": false,
"max_retry_delay": "42.0 seconds",
"multiple_outputs": false,
@@ -861,6 +890,7 @@
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1025,12 +1055,20 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "EmptyOperator",
"operator_class_path":
"airflow.providers.standard.operators.empty.EmptyOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[{'uri': 'file://host/dir/file.txt',
'extra': {}, 'type': 'asset'}]",
+ "outlets_list": [
+ {
+ "uri": "file://host/dir/file.txt",
+ "extra": {},
+ "type": "asset"
+ }
+ ],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1195,12 +1233,20 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "EmptyOperator",
"operator_class_path":
"airflow.providers.standard.operators.empty.EmptyOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[{'uri': 'file://host/dir/file.txt',
'extra': {}, 'type': 'asset'}]",
+ "outlets_list": [
+ {
+ "uri": "file://host/dir/file.txt",
+ "extra": {},
+ "type": "asset"
+ }
+ ],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1362,6 +1408,7 @@
"external_task_id": "external_task_id",
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"max_active_tis_per_dag": 7,
"max_active_tis_per_dagrun": 2,
@@ -1369,6 +1416,7 @@
"operator_class": "SomeCustomOperator",
"operator_class_path": "{{
result.endswith('openlineage_base_complex_dag.SomeCustomOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "owner3",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1521,6 +1569,7 @@
"external_task_id": "external_task_id",
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"max_active_tis_per_dag": 7,
"max_active_tis_per_dagrun": 2,
@@ -1528,6 +1577,7 @@
"operator_class": "SomeCustomOperator",
"operator_class_path": "{{
result.endswith('openlineage_base_complex_dag.SomeCustomOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "owner3",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1677,11 +1727,13 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": "{{ result == true }}",
"multiple_outputs": false,
"operator_class": "CustomMappedOperator",
"operator_class_path": "{{
result.endswith('openlineage_base_complex_dag.CustomMappedOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -1840,11 +1892,13 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": "{{ result == true }}",
"multiple_outputs": false,
"operator_class": "CustomMappedOperator",
"operator_class_path": "{{
result.endswith('openlineage_base_complex_dag.CustomMappedOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -2003,12 +2057,14 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "EmptyOperator",
"operator_class_path":
"airflow.providers.standard.operators.empty.EmptyOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -2176,12 +2232,14 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "EmptyOperator",
"operator_class_path":
"airflow.providers.standard.operators.empty.EmptyOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -2349,12 +2407,14 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "PythonOperator",
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -2521,12 +2581,14 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "PythonOperator",
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_simple_dag__af2.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_simple_dag__af2.json
index b3ba5e56bcf..884c785dc85 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_simple_dag__af2.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_simple_dag__af2.json
@@ -154,6 +154,7 @@
"executor_config": {},
"ignore_first_depends_on_past": true,
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -162,6 +163,7 @@
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -285,6 +287,7 @@
"executor_config": {},
"ignore_first_depends_on_past": true,
"inlets": "[]",
+ "inlets_list": [],
"is_setup": false,
"is_teardown": false,
"mapped": false,
@@ -293,6 +296,7 @@
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_simple_dag__af3.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_simple_dag__af3.json
index 0652f3d3de8..672e8dfab10 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_simple_dag__af3.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_base_simple_dag__af3.json
@@ -153,12 +153,14 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "PythonOperator",
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -279,12 +281,14 @@
"executor_config": {},
"ignore_first_depends_on_past": false,
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "PythonOperator",
"operator_class_path":
"airflow.providers.standard.operators.python.PythonOperator",
"operator_provider_version": "{{ regex_match(result,
\"^[\\d]+\\.[\\d]+\\.[\\d]+.*\") }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_custom_operator_failure_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_custom_operator_failure_dag.json
index b66a4acae33..da4b7f60472 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_custom_operator_failure_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_custom_operator_failure_dag.json
@@ -87,11 +87,13 @@
"downstream_task_ids": "['empty_success']",
"executor_config": {},
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "FailingOLOperator",
"operator_class_path": "{{
result.endswith('.FailingOLOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "some_owner1",
"priority_weight": 1,
"queue": "{{ result is string }}",
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_manual_lineage_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_manual_lineage_dag.json
index 0a0af4aabb8..452f3551746 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_manual_lineage_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_manual_lineage_dag.json
@@ -63,10 +63,12 @@
"downstream_task_ids": "['datasets_maximal']",
"executor_config": {},
"inlets": "[]",
+ "inlets_list": [],
"multiple_outputs": false,
"operator_class": "{{ result is string }}",
"operator_class_path": "{{ result is string }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -236,10 +238,12 @@
"downstream_task_ids": "['query_minimal']",
"executor_config": {},
"inlets": "[]",
+ "inlets_list": [],
"multiple_outputs": false,
"operator_class": "{{ result is string }}",
"operator_class_path": "{{ result is string }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
diff --git
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_taskflow_simple_dag.json
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_taskflow_simple_dag.json
index 2c7ebed8ec1..88616cab2f6 100644
---
a/providers/openlineage/tests/system/openlineage/expected_events/openlineage_taskflow_simple_dag.json
+++
b/providers/openlineage/tests/system/openlineage/expected_events/openlineage_taskflow_simple_dag.json
@@ -148,11 +148,13 @@
"downstream_task_ids": "['check_events']",
"executor_config": {},
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "_PythonDecoratedOperator",
"operator_class_path": "{{
result.endswith('decorators.python._PythonDecoratedOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
@@ -264,11 +266,13 @@
"downstream_task_ids": "['check_events']",
"executor_config": {},
"inlets": "[]",
+ "inlets_list": [],
"mapped": false,
"multiple_outputs": false,
"operator_class": "_PythonDecoratedOperator",
"operator_class_path": "{{
result.endswith('decorators.python._PythonDecoratedOperator') }}",
"outlets": "[]",
+ "outlets_list": [],
"owner": "airflow",
"priority_weight": 1,
"queue": "{{ result is string }}",
diff --git a/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
b/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
index 4078993937e..1b66b80ab5a 100644
--- a/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
+++ b/providers/openlineage/tests/unit/openlineage/utils/test_utils.py
@@ -18,7 +18,9 @@
from __future__ import annotations
import datetime
+import enum
import pathlib
+from decimal import Decimal
from types import SimpleNamespace
from unittest import mock
from unittest.mock import MagicMock, PropertyMock, patch
@@ -33,7 +35,7 @@ from uuid6 import uuid7
from airflow import DAG
from airflow.models.dagrun import DagRun
from airflow.models.taskinstance import TaskInstance, TaskInstanceState
-from airflow.providers.common.compat.assets import Asset
+from airflow.providers.common.compat.assets import Asset, AssetAlias
from airflow.providers.common.compat.sdk import BaseOperator, TaskGroup, task,
timezone
from airflow.providers.openlineage.conf import namespace
from airflow.providers.openlineage.plugins.facets import AirflowDagRunFacet,
AirflowJobFacet
@@ -49,6 +51,7 @@ from airflow.providers.openlineage.utils.utils import (
TaskInstanceInfo,
TaskInstancePosition,
_build_labeled_edge_map,
+ _convert_to_json_safe_list,
_extract_ol_info_from_asset_event,
_get_ol_job_dependencies_from_asset_events,
_get_openlineage_data_from_dagrun_conf,
@@ -3400,6 +3403,7 @@ def test_task_info_af3():
"hitl_summary": "hitl_summary",
"ignore_first_depends_on_past": False,
"inlets": "[{'uri': 'uri1', 'extra': {'a': 1}, 'type': 'asset'}]",
+ "inlets_list": [{"uri": "uri1", "extra": {"a": 1}, "type": "asset"}],
"mapped": False,
"max_active_tis_per_dag": None,
"max_active_tis_per_dagrun": None,
@@ -3410,6 +3414,10 @@ def test_task_info_af3():
"operator_class_path": get_fully_qualified_class_name(task_10),
"operator_provider_version": None, # Custom operator doesn't have
provider version
"outlets": "[{'uri': 'uri2', 'extra': {'b': 2}, 'type': 'asset'},
{'uri': 'uri3', 'extra': {'c': 3}, 'type': 'asset'}]",
+ "outlets_list": [
+ {"uri": "uri2", "extra": {"b": 2}, "type": "asset"},
+ {"uri": "uri3", "extra": {"c": 3}, "type": "asset"},
+ ],
"owner": "airflow",
"priority_weight": 1,
"queue": "default",
@@ -3532,6 +3540,7 @@ def test_task_info_af2():
"is_teardown": False,
"sla": None,
"inlets": "[{'uri': 'uri1', 'extra': {'a': 1}, 'type': 'asset'}]",
+ "inlets_list": [{"uri": "uri1", "extra": {"a": 1}, "type": "asset"}],
"mapped": False,
"max_active_tis_per_dag": None,
"max_active_tis_per_dagrun": None,
@@ -3541,6 +3550,10 @@ def test_task_info_af2():
"operator_class_path": get_fully_qualified_class_name(task_10),
"operator_provider_version": None, # Custom operator doesn't have
provider version
"outlets": "[{'uri': 'uri2', 'extra': {'b': 2}, 'type': 'asset'},
{'uri': 'uri3', 'extra': {'c': 3}, 'type': 'asset'}]",
+ "outlets_list": [
+ {"uri": "uri2", "extra": {"b": 2}, "type": "asset"},
+ {"uri": "uri3", "extra": {"c": 3}, "type": "asset"},
+ ],
"owner": "airflow",
"priority_weight": 1,
"queue": "default",
@@ -5235,6 +5248,93 @@ class TestTaskInfoRuntimeAssets:
assert result["outlets"] == "[{'uri': 's3://x/y.txt', 'extra': {'a':
1}, 'type': 'asset_event'}]"
+class TestTaskInfoAssetLists:
+ """``inlets_list`` / ``outlets_list`` carry the same assets as ``inlets``
/ ``outlets`` as real lists."""
+
+ def test_extra_values_that_break_repr_parsing_stay_valid(self):
+ extra = {"etag": '"0x8D0000000000001"', "owner": "O'Brien",
"has_header": True, "note": None}
+ task = BashOperator(
+ task_id="t",
+ bash_command="echo 1",
+ inlets=[Asset(uri="s3://bucket/in.csv", extra={"k": "v"})],
+ outlets=[Asset(uri="s3://bucket/out.csv", extra=extra)],
+ )
+ info = TaskInfo(task)
+ assert info["inlets_list"] == [{"uri": "s3://bucket/in.csv", "extra":
{"k": "v"}, "type": "asset"}]
+ assert info["outlets_list"] == [{"uri": "s3://bucket/out.csv",
"extra": extra, "type": "asset"}]
+
+ def test_no_assets_gives_empty_lists(self):
+ info = TaskInfo(BashOperator(task_id="t", bash_command="echo 1"))
+ assert info["inlets_list"] == []
+ assert info["outlets_list"] == []
+
+ def test_runtime_assets_included(self):
+ task = BashOperator(task_id="t", bash_command="echo 1", outlets=[])
+ dyn = Asset(
+ uri="s3://bucket/dyn.csv", extra={"size": Decimal("10.5"), "day":
datetime.date(2026, 1, 2)}
+ )
+ aliased = Asset(uri="s3://bucket/aliased.csv", extra={"checks":
[{"passed": True}]})
+ info = TaskInfo(task, runtime_assets=[(dyn, None), (aliased,
AssetAlias("my-alias"))])
+ assert info["outlets_list"] == [
+ {
+ "uri": "s3://bucket/dyn.csv",
+ "extra": {"size": "10.5", "day": "2026-01-02"},
+ "type": "asset_event",
+ },
+ {
+ "uri": "s3://bucket/aliased.csv",
+ "extra": {"checks": [{"passed": True}]},
+ "type": "asset_event_from_alias",
+ "source_alias": "my-alias",
+ },
+ ]
+ assert info["inlets_list"] == []
+
+ def test_task_info_complete_includes_lists(self):
+ task = BashOperator(
+ task_id="t", bash_command="echo 1",
outlets=[Asset(uri="s3://bucket/out.csv", extra={"k": "v"})]
+ )
+ assert TaskInfoComplete(task)["outlets_list"] == [
+ {"uri": "s3://bucket/out.csv", "extra": {"k": "v"}, "type":
"asset"}
+ ]
+
+
+class _DatasetName(str, enum.Enum):
+ SYNC = "user_sync"
+
+
[email protected](
+ ("value", "expected"),
+ [
+ pytest.param([], [], id="empty"),
+ pytest.param(
+ [{("tuple", "key"): 1, "uri": "s3://b/o"}], [{"uri": "s3://b/o"}],
id="non_str_key_dropped"
+ ),
+ pytest.param([{"uri": _DatasetName.SYNC}], [{"uri": "user_sync"}],
id="str_enum_uses_value"),
+ pytest.param([{"n": Decimal("1.5")}], [{"n": "1.5"}], id="decimal"),
+ pytest.param(
+ [{"ts": datetime.datetime(2026, 1, 2, 3, 4, 5)}], [{"ts":
"2026-01-02 03:04:05"}], id="datetime"
+ ),
+ pytest.param(
+ [{"nested": {"items": ["a", None, 1.0]}}], [{"nested": {"items":
["a", None, 1.0]}}], id="nested"
+ ),
+ ],
+)
+def test_convert_to_json_safe_list(value, expected):
+ assert _convert_to_json_safe_list(value, "outlets") == expected
+
+
+@patch("airflow.providers.openlineage.utils.utils.log", autospec=True)
+def test_convert_to_json_safe_list_returns_none_on_failure(mock_log):
+ circular: list = []
+ circular.append(circular)
+ assert _convert_to_json_safe_list(circular, "outlets") is None
+ mock_log.warning.assert_called_once_with(
+ "Unable to serialize task `%s` to a JSON list: %s", "outlets", mock.ANY
+ )
+ assert str(mock_log.warning.call_args.args[2]) == "Circular reference
detected"
+
+
@patch("airflow.providers.openlineage.utils.utils.get_runtime_outlet_assets")
@patch("airflow.providers.openlineage.conf.include_full_task_info")
def test_get_airflow_run_facet_plumbs_runtime_assets(mock_include_full,
mock_get_runtime):
