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 68a137d7d053b..236d854ff0988 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 @@ def __init__( 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 @@ def __init__( 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 @@ def __init__( 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 @@ def _handle_job_error(job: BigQueryJob | UnknownJob) -> None: 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=2, + ) + 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 b71456dd658c8..18007fff91a4b 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 @@ def test_schema_obj_external_table_should_execute_successfully(self, bq_hook, gc ) 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 @@ def test_external_table_should_accept_parquet_format_and_options(self, hook): 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 @@ def test_without_external_table_should_accept_parquet_format_and_options(self, h ] 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 @@ def test_src_fmt_configs_emits_deprecation_warning(self, hook): 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 @@ def test_src_fmt_configs_and_extra_config_both_applied_with_precedence(self, hoo 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 e6869670a5bea..db2e4ee6434bf 100644 --- a/scripts/ci/prek/validate_operators_init_exemptions.txt +++ b/scripts/ci/prek/validate_operators_init_exemptions.txt @@ -19,7 +19,6 @@ 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/google/src/airflow/providers/google/marketing_platform/operators/campaign_manager.py::GoogleCampaignManagerDeleteReportOperator providers/microsoft/azure/src/airflow/providers/microsoft/azure/transfers/gcs_to_wasb.py::GCSToAzureBlobStorageOperator providers/microsoft/psrp/src/airflow/providers/microsoft/psrp/operators/psrp.py::PsrpOperator