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 87acced7347 Validate DatabricksCopyIntoOperator template fields after
rendering (#70339)
87acced7347 is described below
commit 87acced734755310760539c22b43e26e9121e81e
Author: Stefan Wang <[email protected]>
AuthorDate: Fri Jul 24 01:02:00 2026 -0700
Validate DatabricksCopyIntoOperator template fields after rendering (#70339)
* Validate DatabricksCopyIntoOperator template fields after rendering
files, table_name and file_location are template fields, rendered after
__init__
runs. The constructor checked files/pattern mutual exclusivity and rejected
empty
table_name/file_location there, acting on the un-rendered Jinja
expressions. Move
those three checks into execute(); the file_format check reads no template
field
and stays in __init__.
related: #70296
---
.../databricks/operators/databricks_sql.py | 13 ++++---
.../databricks/operators/test_databricks_copy.py | 43 ++++++++++++----------
.../ci/prek/validate_operators_init_exemptions.txt | 1 -
3 files changed, 30 insertions(+), 27 deletions(-)
diff --git
a/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py
b/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py
index f025bc09622..f44278982aa 100644
---
a/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py
+++
b/providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py
@@ -405,12 +405,6 @@ class DatabricksCopyIntoOperator(BaseOperator):
) -> None:
"""Create a new ``DatabricksCopyIntoOperator``."""
super().__init__(**kwargs)
- if files is not None and pattern is not None:
- raise AirflowException("Only one of 'pattern' or 'files' should be
specified")
- if table_name == "":
- raise AirflowException("table_name shouldn't be empty")
- if file_location == "":
- raise AirflowException("file_location shouldn't be empty")
if file_format not in COPY_INTO_APPROVED_FORMATS:
raise AirflowException(f"file_format '{file_format}' isn't
supported")
self.files = files
@@ -540,6 +534,13 @@ FILEFORMAT = {self._file_format}
return build_query_tags(context, self.query_tags,
self.include_airflow_query_tags)
def execute(self, context: Context) -> Any:
+ # files/table_name/file_location are template fields; validate after
rendering.
+ if self.files is not None and self._pattern is not None:
+ raise AirflowException("Only one of 'pattern' or 'files' should be
specified")
+ if self.table_name == "":
+ raise AirflowException("table_name shouldn't be empty")
+ if self.file_location == "":
+ raise AirflowException("file_location shouldn't be empty")
self._sql = self._create_sql_query()
self.log.info("Executing: %s", self._sql)
hook = self._get_hook()
diff --git
a/providers/databricks/tests/unit/databricks/operators/test_databricks_copy.py
b/providers/databricks/tests/unit/databricks/operators/test_databricks_copy.py
index f00653f8b22..1d369a3343b 100644
---
a/providers/databricks/tests/unit/databricks/operators/test_databricks_copy.py
+++
b/providers/databricks/tests/unit/databricks/operators/test_databricks_copy.py
@@ -195,37 +195,40 @@ VALIDATE 10 ROWS
def test_incorrect_params_files_patterns():
exception_message = "Only one of 'pattern' or 'files' should be specified"
+ op = DatabricksCopyIntoOperator(
+ task_id=TASK_ID,
+ file_location=COPY_FILE_LOCATION,
+ file_format="JSON",
+ table_name="test",
+ files=["file1", "file2", "file3"],
+ pattern="abc",
+ )
with pytest.raises(AirflowException, match=exception_message):
- DatabricksCopyIntoOperator(
- task_id=TASK_ID,
- file_location=COPY_FILE_LOCATION,
- file_format="JSON",
- table_name="test",
- files=["file1", "file2", "file3"],
- pattern="abc",
- )
+ op.execute(context={})
def test_incorrect_params_emtpy_table():
exception_message = "table_name shouldn't be empty"
+ op = DatabricksCopyIntoOperator(
+ task_id=TASK_ID,
+ file_location=COPY_FILE_LOCATION,
+ file_format="JSON",
+ table_name="",
+ )
with pytest.raises(AirflowException, match=exception_message):
- DatabricksCopyIntoOperator(
- task_id=TASK_ID,
- file_location=COPY_FILE_LOCATION,
- file_format="JSON",
- table_name="",
- )
+ op.execute(context={})
def test_incorrect_params_emtpy_location():
exception_message = "file_location shouldn't be empty"
+ op = DatabricksCopyIntoOperator(
+ task_id=TASK_ID,
+ file_location="",
+ file_format="JSON",
+ table_name="abc",
+ )
with pytest.raises(AirflowException, match=exception_message):
- DatabricksCopyIntoOperator(
- task_id=TASK_ID,
- file_location="",
- file_format="JSON",
- table_name="abc",
- )
+ op.execute(context={})
def test_incorrect_params_wrong_format():
diff --git a/scripts/ci/prek/validate_operators_init_exemptions.txt
b/scripts/ci/prek/validate_operators_init_exemptions.txt
index 881d2cd90d1..052da40bfaa 100644
--- a/scripts/ci/prek/validate_operators_init_exemptions.txt
+++ b/scripts/ci/prek/validate_operators_init_exemptions.txt
@@ -31,7 +31,6 @@
providers/cohere/src/airflow/providers/cohere/operators/embedding.py::CohereEmbe
providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py::DatabricksReposCreateOperator
providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py::DatabricksReposDeleteOperator
providers/databricks/src/airflow/providers/databricks/operators/databricks_repos.py::DatabricksReposUpdateOperator
-providers/databricks/src/airflow/providers/databricks/operators/databricks_sql.py::DatabricksCopyIntoOperator
providers/databricks/src/airflow/providers/databricks/sensors/databricks.py::DatabricksSQLStatementsSensor
providers/docker/src/airflow/providers/docker/operators/docker.py::DockerOperator
providers/google/src/airflow/providers/google/cloud/operators/bigquery.py::BigQueryInsertJobOperator
