nit: Switch to using json instead of str when casting collections in OpenLineage

providers/openlineage/src/airflow/providers/openlineage/utils/utils.py

@@ -298,7 +298,7 @@ def _cast_basic_types(value):
         if isinstance(value, datetime.timedelta):
             return f"{value.total_seconds()} seconds"
         if isinstance(value, (set, list, tuple)):
-            return str(list(value))
+            return json.dumps(list(value))
         return value
 
     def _rename_fields(self):

providers/openlineage/tests/unit/openlineage/plugins/test_utils.py

@@ -208,7 +208,7 @@ class Test:
 
     obj = Test(["a", "b", "c"])
 
-    assert json.loads(json.dumps(TestInfo(obj))) == {"alist": "['a', 'b', 'c']"}
+    assert json.loads(json.dumps(TestInfo(obj))) == {"alist": '["a", "b", "c"]'}
 
 
 def test_info_json_encodable_list_does_include_nonexisting():

providers/openlineage/tests/unit/openlineage/utils/test_utils.py

@@ -176,7 +176,7 @@ def test_get_airflow_dag_run_facet():
         "owner": "airflow",
         "timetable": {},
         "start_date": "2024-06-01T00:00:00+00:00",
-        "tags": "['test']",
+        "tags": '["test"]',
         "owner_links": {},
     }
     if hasattr(dag, "schedule_interval"):  # Airflow 2 compat.
@@ -823,7 +823,7 @@ def test_dag_info(self):
             "fileloc": pathlib.Path(__file__).resolve().as_posix(),
             "schedule_interval": "@once",
             "start_date": "2024-06-01T00:00:00+00:00",
-            "tags": "['test']",
+            "tags": '["test"]',
             "timetable": {},
             "owner_links": {"some_owner": "https://airflow.apache.org"},
         }
@@ -1071,7 +1071,7 @@ def test_dag_info(self):
             "description": "test desc",
             "fileloc": pathlib.Path(__file__).resolve().as_posix(),
             "start_date": "2024-06-01T00:00:00+00:00",
-            "tags": "['test']",
+            "tags": '["test"]',
             "timetable": {},
             "timetable_summary": "@once",
             "owner_links": {"some_owner": "https://airflow.apache.org"},
@@ -1562,7 +1562,7 @@ def __init__(self, *args, **kwargs):
     tg_info = TaskGroupInfo(tg)
     assert dict(tg_info) == {
         "downstream_group_ids": "[]",
-        "downstream_task_ids": "['task_1']",
+        "downstream_task_ids": '["task_1"]',
         "group_id": "section_1",
         "prefix_group_id": True,
         "tooltip": "",
@@ -1572,21 +1572,21 @@ def __init__(self, *args, **kwargs):
     assert dict(result) == {
         "deferrable": True,
         "depends_on_past": False,
-        "downstream_task_ids": "['task_1']",
+        "downstream_task_ids": '["task_1"]',
         "execution_timeout": None,
         "executor_config": {},
         "external_dag_id": "external_dag_id",
         "external_task_id": "external_task_id",
         "ignore_first_depends_on_past": False,
-        "inlets": "[{'uri': 'uri1', 'extra': {'a': 1}}]",
+        "inlets": '[{"uri": "uri1", "extra": {"a": 1}}]',
         "mapped": False,
         "max_active_tis_per_dag": None,
         "max_active_tis_per_dagrun": None,
         "max_retry_delay": None,
         "multiple_outputs": False,
         "operator_class": "CustomOperator",
         "operator_class_path": get_fully_qualified_class_name(task_10),
-        "outlets": "[{'uri': 'uri2', 'extra': {'b': 2}}, {'uri': 'uri3', 'extra': {'c': 3}}]",
+        "outlets": '[{"uri": "uri2", "extra": {"b": 2}}, {"uri": "uri3", "extra": {"c": 3}}]',
         "owner": "airflow",
         "priority_weight": 1,
         "queue": "default",
@@ -1597,7 +1597,7 @@ def __init__(self, *args, **kwargs):
         "task_id": "section_1.task_3",
         "trigger_dag_id": "trigger_dag_id",
         "trigger_rule": "all_success",
-        "upstream_task_ids": "['task_0']",
+        "upstream_task_ids": '["task_0"]',
         "wait_for_downstream": False,
         "wait_for_past_depends_before_skipping": False,
     }
@@ -1636,7 +1636,7 @@ def __init__(self, *args, **kwargs):
     tg_info = TaskGroupInfo(tg)
     assert dict(tg_info) == {
         "downstream_group_ids": "[]",
-        "downstream_task_ids": "['task_1']",
+        "downstream_task_ids": '["task_1"]',
         "group_id": "section_1",
         "prefix_group_id": True,
         "tooltip": "",
@@ -1646,7 +1646,7 @@ def __init__(self, *args, **kwargs):
     assert dict(result) == {
         "deferrable": True,
         "depends_on_past": False,
-        "downstream_task_ids": "['task_1']",
+        "downstream_task_ids": '["task_1"]',
         "execution_timeout": None,
         "executor_config": {},
         "external_dag_id": "external_dag_id",
@@ -1655,15 +1655,15 @@ def __init__(self, *args, **kwargs):
         "is_setup": False,
         "is_teardown": False,
         "sla": None,
-        "inlets": "[{'uri': 'uri1', 'extra': {'a': 1}}]",
+        "inlets": '[{"uri": "uri1", "extra": {"a": 1}}]',
         "mapped": False,
         "max_active_tis_per_dag": None,
         "max_active_tis_per_dagrun": None,
         "max_retry_delay": None,
         "multiple_outputs": False,
         "operator_class": "CustomOperator",
         "operator_class_path": get_fully_qualified_class_name(task_10),
-        "outlets": "[{'uri': 'uri2', 'extra': {'b': 2}}, {'uri': 'uri3', 'extra': {'c': 3}}]",
+        "outlets": '[{"uri": "uri2", "extra": {"b": 2}}, {"uri": "uri3", "extra": {"c": 3}}]',
         "owner": "airflow",
         "priority_weight": 1,
         "queue": "default",
@@ -1674,7 +1674,7 @@ def __init__(self, *args, **kwargs):
         "task_id": "section_1.task_3",
         "trigger_dag_id": "trigger_dag_id",
         "trigger_rule": "all_success",
-        "upstream_task_ids": "['task_0']",
+        "upstream_task_ids": '["task_0"]',
         "wait_for_downstream": False,
         "wait_for_past_depends_before_skipping": False,
     }