This is an automated email from the ASF dual-hosted git repository.
potiuk 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 d450613b2eb Do not allow semicolons in
CopyFromExternalStageToSnowflakeOperator fieldS (#51734)
d450613b2eb is described below
commit d450613b2ebfa62a1edc470256c38c843aebd5cb
Author: Amogh Desai <[email protected]>
AuthorDate: Sat Jun 14 22:04:35 2025 +0530
Do not allow semicolons in CopyFromExternalStageToSnowflakeOperator fieldS
(#51734)
---
.../snowflake/transfers/copy_into_snowflake.py | 15 +++++++++++---
.../transfers/test_copy_into_snowflake.py | 23 ++++++++++++++++++++++
2 files changed, 35 insertions(+), 3 deletions(-)
diff --git
a/providers/snowflake/src/airflow/providers/snowflake/transfers/copy_into_snowflake.py
b/providers/snowflake/src/airflow/providers/snowflake/transfers/copy_into_snowflake.py
index 0a29ecd9974..e8b70344fbf 100644
---
a/providers/snowflake/src/airflow/providers/snowflake/transfers/copy_into_snowflake.py
+++
b/providers/snowflake/src/airflow/providers/snowflake/transfers/copy_into_snowflake.py
@@ -27,6 +27,15 @@ from airflow.providers.snowflake.hooks.snowflake import
SnowflakeHook
from airflow.providers.snowflake.utils.common import enclose_param
+def _validate_parameter(param_name: str, value: str | None) -> str | None:
+ """Validate that the parameter doesn't contain any invalid pattern."""
+ if value is None:
+ return None
+ if ";" in value:
+ raise ValueError(f"Invalid {param_name}: semicolons (;) not allowed.")
+ return value
+
+
class CopyFromExternalStageToSnowflakeOperator(BaseOperator):
"""
Executes a COPY INTO command to load files from an external stage from
clouds to Snowflake.
@@ -91,8 +100,8 @@ class CopyFromExternalStageToSnowflakeOperator(BaseOperator):
):
super().__init__(**kwargs)
self.files = files
- self.table = table
- self.stage = stage
+ self.table = _validate_parameter("table", table)
+ self.stage = _validate_parameter("stage", stage)
self.prefix = prefix
self.file_format = file_format
self.schema = schema
@@ -126,7 +135,7 @@ class
CopyFromExternalStageToSnowflakeOperator(BaseOperator):
if self.schema:
into = f"{self.schema}.{self.table}"
else:
- into = self.table
+ into = self.table # type: ignore[assignment]
if self.columns_array:
into = f"{into}({', '.join(self.columns_array)})"
diff --git
a/providers/snowflake/tests/unit/snowflake/transfers/test_copy_into_snowflake.py
b/providers/snowflake/tests/unit/snowflake/transfers/test_copy_into_snowflake.py
index 5b0bf27f600..ed0bd3b2dc7 100644
---
a/providers/snowflake/tests/unit/snowflake/transfers/test_copy_into_snowflake.py
+++
b/providers/snowflake/tests/unit/snowflake/transfers/test_copy_into_snowflake.py
@@ -549,3 +549,26 @@ class TestCopyFromExternalStageToSnowflake:
},
job_facets={"sql": SQLJobFacet(query=expected_sql)},
)
+
+ def test_init_with_invalid_parameters(self):
+ # Test by passing invalid parameters to the operator
+ import re
+
+ with pytest.raises(ValueError, match=re.escape("Invalid table:
semicolons (;) not allowed.")):
+ _ = CopyFromExternalStageToSnowflakeOperator(
+ task_id="test",
+ table="table; some_new_table",
+ stage="stage",
+ database="",
+ schema="schema",
+ file_format="CSV",
+ )
+ with pytest.raises(ValueError, match=re.escape("Invalid stage:
semicolons (;) not allowed.")):
+ _ = CopyFromExternalStageToSnowflakeOperator(
+ task_id="test",
+ table="table",
+ stage="stage; some_new_stabge",
+ database="",
+ schema="schema",
+ file_format="CSV",
+ )
