This is an automated email from the ASF dual-hosted git repository.
shahar1 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 734e8b3d70d Emit GCSToBigQueryOperator deprecation warning after
rendering (#70542)
734e8b3d70d is described below
commit 734e8b3d70d909bab59b21a90c692a28fadfc17d
Author: Yuseok Jo <[email protected]>
AuthorDate: Sat Sep 12 18:16:45 2026 +0900
Emit GCSToBigQueryOperator deprecation warning after rendering (#70542)
---
.../google/cloud/transfers/gcs_to_bigquery.py | 38 +++---
.../google/cloud/transfers/test_gcs_to_bigquery.py | 135 ++++++++++++---------
.../ci/prek/validate_operators_init_exemptions.txt | 1 -
3 files changed, 104 insertions(+), 70 deletions(-)
diff --git
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
index 68a137d7d05..549f405e0fa 100644
---
a/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
+++
b/providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py
@@ -261,8 +261,6 @@ class GCSToBigQueryOperator(BaseOperator):
self.configuration: dict[str, Any] = {}
# GCS config
- if src_fmt_configs is None:
- src_fmt_configs = {}
if time_partitioning is None:
time_partitioning = {}
if range_partitioning is None:
@@ -272,9 +270,6 @@ class GCSToBigQueryOperator(BaseOperator):
self.bucket = bucket
self.source_objects = source_objects
self.schema_object = schema_object
-
- if schema_object_bucket is None:
- schema_object_bucket = bucket
self.schema_object_bucket = schema_object_bucket
# BQ config
@@ -305,16 +300,6 @@ class GCSToBigQueryOperator(BaseOperator):
self.schema_update_options = schema_update_options
self.src_fmt_configs = src_fmt_configs
- if src_fmt_configs:
- warnings.warn(
- "The 'src_fmt_configs' parameter is deprecated. Use
'extra_config' instead. "
- "Note: 'extra_config' uses the fully-nested API structure, so
format-specific "
- "options must be nested under their parent key "
- "(e.g., {'parquetOptions': {'enableListInference': True}}
rather than "
- "{'enableListInference': True}).",
- AirflowProviderDeprecationWarning,
- stacklevel=2,
- )
self.extra_config = extra_config
self.time_partitioning = time_partitioning
self.range_partitioning = range_partitioning
@@ -359,7 +344,30 @@ class GCSToBigQueryOperator(BaseOperator):
if job.error_result:
raise AirflowException(f"BigQuery job {job.job_id} failed:
{job.error_result}")
+ def _warn_on_deprecated_template_fields(self) -> None:
+ if self.src_fmt_configs:
+ warnings.warn(
+ "The 'src_fmt_configs' parameter is deprecated. Use
'extra_config' instead. "
+ "Note: 'extra_config' uses the fully-nested API structure, so
format-specific "
+ "options must be nested under their parent key "
+ "(e.g., {'parquetOptions': {'enableListInference': True}}
rather than "
+ "{'enableListInference': True}).",
+ AirflowProviderDeprecationWarning,
+ stacklevel=1,
+ )
+
def execute(self, context: Context):
+ # Template fields render after __init__, so defaults that depend on a
template field
+ # (schema_object_bucket falls back to bucket) and the src_fmt_configs
deprecation check
+ # must run here, against the rendered values.
+ if self.src_fmt_configs is None:
+ self.src_fmt_configs = {}
+ if self.schema_object_bucket is None:
+ self.schema_object_bucket = self.bucket
+ # Not captured in the rendered-template view (it defaults after
rendering), so log it.
+ self.log.info("schema_object_bucket not set, defaulting to bucket
%s", self.bucket)
+ self._warn_on_deprecated_template_fields()
+
hook = BigQueryHook(
gcp_conn_id=self.gcp_conn_id,
location=self.location,
diff --git
a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
index b71456dd658..18007fff91a 100644
--- a/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
+++ b/providers/google/tests/unit/google/cloud/transfers/test_gcs_to_bigquery.py
@@ -812,6 +812,33 @@ class TestGCSToBigQueryOperator:
)
gcs_hook.return_value.download.assert_called_once_with(SCHEMA_BUCKET,
SCHEMA_OBJECT)
+ @mock.patch(GCS_TO_BQ_PATH.format("GCSHook"))
+ @mock.patch(GCS_TO_BQ_PATH.format("BigQueryHook"))
+ def test_schema_object_bucket_defaults_to_bucket_when_omitted(self,
bq_hook, gcs_hook):
+ # The schema_object_bucket -> bucket fallback now runs in execute()
after rendering; when
+ # schema_object_bucket is omitted the schema download must still
target the bucket.
+ bq_hook.return_value.insert_job.side_effect = [
+ MagicMock(job_id=REAL_JOB_ID, error_result=False),
+ REAL_JOB_ID,
+ ]
+ bq_hook.return_value.generate_job_id.return_value = REAL_JOB_ID
+ bq_hook.return_value.split_tablename.return_value = (PROJECT_ID,
DATASET, TABLE)
+ gcs_hook.return_value.download.return_value =
bytes(json.dumps(SCHEMA_FIELDS), "utf-8")
+ operator = GCSToBigQueryOperator(
+ task_id=TASK_ID,
+ bucket=TEST_BUCKET,
+ source_objects=TEST_SOURCE_OBJECTS,
+ schema_object=SCHEMA_OBJECT,
+ write_disposition=WRITE_DISPOSITION,
+ destination_project_dataset_table=TEST_EXPLICIT_DEST,
+ external_table=True,
+ project_id=JOB_PROJECT_ID,
+ )
+
+ operator.execute(context=MagicMock())
+
+ gcs_hook.return_value.download.assert_called_once_with(TEST_BUCKET,
SCHEMA_OBJECT)
+
@mock.patch(GCS_TO_BQ_PATH.format("GCSHook"))
@mock.patch(GCS_TO_BQ_PATH.format("BigQueryHook"))
def
test_schema_obj_without_external_table_should_execute_successfully(self,
bq_hook, gcs_hook):
@@ -1742,23 +1769,23 @@ class TestGCSToBigQueryOperator:
hook.return_value.generate_job_id.return_value = REAL_JOB_ID
hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET,
TABLE)
- with pytest.warns(AirflowProviderDeprecationWarning,
match="src_fmt_configs"):
- operator = GCSToBigQueryOperator(
- task_id=TASK_ID,
- bucket=TEST_BUCKET,
- source_objects=TEST_SOURCE_OBJECTS,
- destination_project_dataset_table=TEST_EXPLICIT_DEST,
- schema_fields=SCHEMA_FIELDS,
- write_disposition=WRITE_DISPOSITION,
- external_table=True,
- project_id=JOB_PROJECT_ID,
- source_format="PARQUET",
- src_fmt_configs={
- "enableListInference": True,
- },
- )
+ operator = GCSToBigQueryOperator(
+ task_id=TASK_ID,
+ bucket=TEST_BUCKET,
+ source_objects=TEST_SOURCE_OBJECTS,
+ destination_project_dataset_table=TEST_EXPLICIT_DEST,
+ schema_fields=SCHEMA_FIELDS,
+ write_disposition=WRITE_DISPOSITION,
+ external_table=True,
+ project_id=JOB_PROJECT_ID,
+ source_format="PARQUET",
+ src_fmt_configs={
+ "enableListInference": True,
+ },
+ )
- operator.execute(context=MagicMock())
+ with pytest.warns(AirflowProviderDeprecationWarning,
match="src_fmt_configs"):
+ operator.execute(context=MagicMock())
hook.return_value.create_table.assert_called_once_with(
exists_ok=True,
@@ -1845,22 +1872,22 @@ class TestGCSToBigQueryOperator:
]
hook.return_value.generate_job_id.return_value = REAL_JOB_ID
hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET,
TABLE)
- with pytest.warns(AirflowProviderDeprecationWarning,
match="src_fmt_configs"):
- operator = GCSToBigQueryOperator(
- task_id=TASK_ID,
- bucket=TEST_BUCKET,
- source_objects=TEST_SOURCE_OBJECTS,
- write_disposition=WRITE_DISPOSITION,
- destination_project_dataset_table=TEST_EXPLICIT_DEST,
- external_table=False,
- project_id=JOB_PROJECT_ID,
- source_format="PARQUET",
- src_fmt_configs={
- "enableListInference": True,
- },
- )
+ operator = GCSToBigQueryOperator(
+ task_id=TASK_ID,
+ bucket=TEST_BUCKET,
+ source_objects=TEST_SOURCE_OBJECTS,
+ write_disposition=WRITE_DISPOSITION,
+ destination_project_dataset_table=TEST_EXPLICIT_DEST,
+ external_table=False,
+ project_id=JOB_PROJECT_ID,
+ source_format="PARQUET",
+ src_fmt_configs={
+ "enableListInference": True,
+ },
+ )
- operator.execute(context=MagicMock())
+ with pytest.warns(AirflowProviderDeprecationWarning,
match="src_fmt_configs"):
+ operator.execute(context=MagicMock())
calls = [
call(
@@ -2054,18 +2081,18 @@ class TestGCSToBigQueryOperator:
hook.return_value.generate_job_id.return_value = REAL_JOB_ID
hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET,
TABLE)
- with pytest.warns(AirflowProviderDeprecationWarning,
match="src_fmt_configs"):
- operator = GCSToBigQueryOperator(
- task_id=TASK_ID,
- bucket=TEST_BUCKET,
- source_objects=TEST_SOURCE_OBJECTS,
- destination_project_dataset_table=TEST_EXPLICIT_DEST,
- write_disposition=WRITE_DISPOSITION,
- project_id=JOB_PROJECT_ID,
- src_fmt_configs={"skipLeadingRows": 1},
- )
+ operator = GCSToBigQueryOperator(
+ task_id=TASK_ID,
+ bucket=TEST_BUCKET,
+ source_objects=TEST_SOURCE_OBJECTS,
+ destination_project_dataset_table=TEST_EXPLICIT_DEST,
+ write_disposition=WRITE_DISPOSITION,
+ project_id=JOB_PROJECT_ID,
+ src_fmt_configs={"skipLeadingRows": 1},
+ )
- operator.execute(context=MagicMock())
+ with pytest.warns(AirflowProviderDeprecationWarning,
match="src_fmt_configs"):
+ operator.execute(context=MagicMock())
config = hook.return_value.insert_job.call_args[1]["configuration"]
assert config["load"]["skipLeadingRows"] == 1
@@ -2076,19 +2103,19 @@ class TestGCSToBigQueryOperator:
hook.return_value.generate_job_id.return_value = REAL_JOB_ID
hook.return_value.split_tablename.return_value = (PROJECT_ID, DATASET,
TABLE)
- with pytest.warns(AirflowProviderDeprecationWarning,
match="src_fmt_configs"):
- operator = GCSToBigQueryOperator(
- task_id=TASK_ID,
- bucket=TEST_BUCKET,
- source_objects=TEST_SOURCE_OBJECTS,
- destination_project_dataset_table=TEST_EXPLICIT_DEST,
- write_disposition=WRITE_DISPOSITION,
- project_id=JOB_PROJECT_ID,
- src_fmt_configs={"skipLeadingRows": 1},
- extra_config={"skipLeadingRows": 5, "columnNameCharacterMap":
"STRICT"},
- )
+ operator = GCSToBigQueryOperator(
+ task_id=TASK_ID,
+ bucket=TEST_BUCKET,
+ source_objects=TEST_SOURCE_OBJECTS,
+ destination_project_dataset_table=TEST_EXPLICIT_DEST,
+ write_disposition=WRITE_DISPOSITION,
+ project_id=JOB_PROJECT_ID,
+ src_fmt_configs={"skipLeadingRows": 1},
+ extra_config={"skipLeadingRows": 5, "columnNameCharacterMap":
"STRICT"},
+ )
- operator.execute(context=MagicMock())
+ with pytest.warns(AirflowProviderDeprecationWarning,
match="src_fmt_configs"):
+ operator.execute(context=MagicMock())
config = hook.return_value.insert_job.call_args[1]["configuration"]
# extra_config wins for overlapping key
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt
b/scripts/ci/prek/validate_operators_init_exemptions.txt
index af75c526a1d..25cb0fe1a14 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -18,5 +18,4 @@
providers/google/src/airflow/providers/google/cloud/operators/gcs.py::GCSFileTra
providers/google/src/airflow/providers/google/cloud/sensors/bigquery_dts.py::BigQueryDataTransferServiceTransferRunSensor
providers/google/src/airflow/providers/google/cloud/sensors/cloud_composer.py::CloudComposerExternalTaskSensor
providers/google/src/airflow/providers/google/cloud/transfers/azure_fileshare_to_gcs.py::AzureFileShareToGCSOperator
-providers/google/src/airflow/providers/google/cloud/transfers/gcs_to_bigquery.py::GCSToBigQueryOperator
providers/microsoft/psrp/src/airflow/providers/microsoft/psrp/operators/psrp.py::PsrpOperator
