Skip to content

fix: revert behavior of flattening lists in OpenLineage's InfoJsonEncodable #41786

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Merged
mobuchowski merged 1 commit into from
Sep 2, 2024
Merged

fix: revert behavior of flattening lists in OpenLineage's InfoJsonEncodable #41786

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
2 changes: 1 addition & 1 deletion airflow/providers/openlineage/utils/utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -199,7 +199,7 @@ def _cast_basic_types(value):
return value.isoformat()
if isinstance(value, datetime.timedelta):
return f"{value.total_seconds()} seconds"
if isinstance(value, (set, tuple)):
if isinstance(value, (set, list, tuple)):
return str(list(value))
return value

Expand Down
99 changes: 48 additions & 51 deletions tests/providers/openlineage/plugins/test_adapter.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -528,7 +528,7 @@ def test_emit_failed_event_with_additional_information(mock_stats_incr, mock_sta
mock_stats_timer.assert_called_with("ol.emit.attempts")


@mock.patch("airflow.providers.openlineage.conf.debug_mode", return_value=True)
@mock.patch("airflow.providers.openlineage.conf.debug_mode", return_value=False)
@mock.patch("airflow.providers.openlineage.plugins.adapter.generate_static_uuid")
@mock.patch("airflow.providers.openlineage.plugins.adapter.Stats.timer")
@mock.patch("airflow.providers.openlineage.plugins.adapter.Stats.incr")
Expand Down Expand Up @@ -578,7 +578,7 @@ def test_emit_dag_started_event(mock_stats_incr, mock_stats_timer, generate_stat
"description": "dag desc",
"owner": "airflow",
"start_date": "2024-06-01T00:00:00+00:00",
"tags": [],
"tags": "[]",
"fileloc": pathlib.Path(__file__).resolve().as_posix(),
}
if hasattr(dag, "schedule_interval"): # Airflow 2 compat.
Expand All @@ -587,56 +587,53 @@ def test_emit_dag_started_event(mock_stats_incr, mock_stats_timer, generate_stat
expected_dag_info["timetable_summary"] = "1 day, 0:00:00"

assert len(client.emit.mock_calls) == 1
assert (
call(
RunEvent(
eventType=RunState.START,
eventTime=event_time.isoformat(),
run=Run(
runId=random_uuid,
facets={
"nominalTime": nominal_time_run.NominalTimeRunFacet(
nominalStartTime=event_time.isoformat(),
nominalEndTime=event_time.isoformat(),
),
"airflowDagRun": AirflowDagRunFacet(
dag=expected_dag_info,
dagRun={
"conf": {},
"dag_id": "dag_id",
"data_interval_start": event_time.isoformat(),
"data_interval_end": event_time.isoformat(),
"external_trigger": None,
"run_id": run_id,
"run_type": None,
"start_date": event_time.isoformat(),
},
),
"debug": AirflowDebugRunFacet(packages=ANY),
},
),
job=Job(
namespace=namespace(),
name="dag_id",
facets={
"documentation": documentation_job.DocumentationJobFacet(description="dag desc"),
"ownership": ownership_job.OwnershipJobFacet(
owners=[
ownership_job.Owner(name="airflow", type=None),
]
),
**job_facets,
"jobType": job_type_job.JobTypeJobFacet(
processingType="BATCH", integration="AIRFLOW", jobType="DAG"
),
},
),
producer=_PRODUCER,
inputs=[],
outputs=[],
)
client.emit.assert_called_once_with(
RunEvent(
eventType=RunState.START,
eventTime=event_time.isoformat(),
run=Run(
runId=random_uuid,
facets={
"nominalTime": nominal_time_run.NominalTimeRunFacet(
nominalStartTime=event_time.isoformat(),
nominalEndTime=event_time.isoformat(),
),
"airflowDagRun": AirflowDagRunFacet(
dag=expected_dag_info,
dagRun={
"conf": {},
"dag_id": "dag_id",
"data_interval_start": event_time.isoformat(),
"data_interval_end": event_time.isoformat(),
"external_trigger": None,
"run_id": run_id,
"run_type": None,
"start_date": event_time.isoformat(),
},
),
# "debug": AirflowDebugRunFacet(packages=ANY),
},
),
job=Job(
namespace=namespace(),
name="dag_id",
facets={
"documentation": documentation_job.DocumentationJobFacet(description="dag desc"),
"ownership": ownership_job.OwnershipJobFacet(
owners=[
ownership_job.Owner(name="airflow", type=None),
]
),
**job_facets,
"jobType": job_type_job.JobTypeJobFacet(
processingType="BATCH", integration="AIRFLOW", jobType="DAG"
),
},
),
producer=_PRODUCER,
inputs=[],
outputs=[],
)
in client.emit.mock_calls
)

mock_stats_incr.assert_not_called()
Expand Down
4 changes: 2 additions & 2 deletions tests/providers/openlineage/plugins/test_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -164,7 +164,7 @@ class Test:
}


def test_info_json_encodable_list_does_not_flatten():
def test_info_json_encodable_list_does_flatten():
class TestInfo(InfoJsonEncodable):
includes = ["alist"]

Expand All @@ -174,7 +174,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():
Expand Down
2 changes: 1 addition & 1 deletion tests/providers/openlineage/utils/test_utils.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -142,7 +142,7 @@ def test_get_airflow_dag_run_facet():
"owner": "airflow",
"timetable": {},
"start_date": "2024-06-01T00:00:00+00:00",
"tags": ["test"],
"tags": "['test']",
}
if hasattr(dag, "schedule_interval"): # Airflow 2 compat.
expected_dag_info["schedule_interval"] = "@once"
Expand Down