You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@airflow.apache.org by po...@apache.org on 2026/05/10 02:13:36 UTC
(airflow) branch main updated: replace AirflowExceptions with native exceptions in Snowflake provider (#66279)
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 4c96fb6f2ec replace AirflowExceptions with native exceptions in Snowflake provider (#66279)
4c96fb6f2ec is described below
commit 4c96fb6f2ecfa3c761b6f709eabe8da52c775c00
Author: Justin Pakzad <11...@users.noreply.github.com>
AuthorDate: Sat May 9 22:13:28 2026 -0400
replace AirflowExceptions with native exceptions in Snowflake provider (#66279)
---
.../src/airflow/providers/snowflake/hooks/snowflake.py | 3 +--
.../airflow/providers/snowflake/hooks/snowflake_sql_api.py | 7 +++----
.../src/airflow/providers/snowflake/operators/snowflake.py | 8 ++++----
.../tests/unit/snowflake/hooks/test_snowflake_sql_api.py | 11 +++++------
.../tests/unit/snowflake/operators/test_snowflake.py | 14 +++++++-------
scripts/ci/prek/known_airflow_exceptions.txt | 4 ----
6 files changed, 20 insertions(+), 27 deletions(-)
diff --git a/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake.py b/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake.py
index 1f9d1b0c5b9..b4a8a91e31d 100644
--- a/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake.py
+++ b/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake.py
@@ -37,7 +37,6 @@ from requests.exceptions import ConnectionError, HTTPError, Timeout
from sqlalchemy import create_engine
from airflow.providers.common.compat.sdk import (
- AirflowException,
AirflowOptionalProviderFeatureException,
Connection,
conf,
@@ -586,7 +585,7 @@ class SnowflakeHook(DbApiHook):
p_key = None
if private_key_content and private_key_file:
- raise AirflowException(
+ raise ValueError(
"The private_key_file and private_key_content extra fields are mutually exclusive. "
"Please remove one."
)
diff --git a/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake_sql_api.py b/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake_sql_api.py
index 5479e5a8727..bd0ea6b6485 100644
--- a/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake_sql_api.py
+++ b/providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake_sql_api.py
@@ -37,7 +37,6 @@ from tenacity import (
)
from airflow.exceptions import AirflowProviderDeprecationWarning
-from airflow.providers.common.compat.sdk import AirflowException
from airflow.providers.common.sql.hooks.lineage import send_sql_hook_lineage
from airflow.providers.snowflake.hooks.snowflake import SnowflakeHook
from airflow.providers.snowflake.utils.sql_api_generate_jwt import JWTGenerator
@@ -161,7 +160,7 @@ class SnowflakeSqlApiHook(SnowflakeHook):
headers = self.get_headers()
sql_is_multi_stmt = ";" in sql.strip()
if not isinstance(bindings, dict) and bindings is not None:
- raise AirflowException("Bindings should be a dictionary or None.")
+ raise TypeError("Bindings should be a dictionary or None.")
if bindings and sql_is_multi_stmt:
self.log.warning(
"Bindings are not supported for multi-statement queries. Bindings will be ignored."
@@ -191,7 +190,7 @@ class SnowflakeSqlApiHook(SnowflakeHook):
elif "statementHandle" in json_response:
self.query_ids.append(json_response["statementHandle"])
else:
- raise AirflowException("No statementHandle/statementHandles present in response")
+ raise RuntimeError("No statementHandle/statementHandles present in response")
# Send Hook Level Lineage
len_query_ids = len(self.query_ids)
@@ -238,7 +237,7 @@ class SnowflakeSqlApiHook(SnowflakeHook):
if conn_config.get("authenticator") == "programmatic_access_token":
pat = conn_config.get("password")
if not pat:
- raise AirflowException(
+ raise ValueError(
"Programmatic Access Token (PAT) authentication requires the connection password "
"field to contain the PAT token value."
)
diff --git a/providers/snowflake/src/airflow/providers/snowflake/operators/snowflake.py b/providers/snowflake/src/airflow/providers/snowflake/operators/snowflake.py
index 0cd5929a1ba..2b02d259e71 100644
--- a/providers/snowflake/src/airflow/providers/snowflake/operators/snowflake.py
+++ b/providers/snowflake/src/airflow/providers/snowflake/operators/snowflake.py
@@ -23,7 +23,7 @@ from datetime import timedelta
from functools import cached_property
from typing import TYPE_CHECKING, Any, SupportsAbs, cast
-from airflow.providers.common.compat.sdk import AirflowException, conf
+from airflow.providers.common.compat.sdk import conf
from airflow.providers.common.sql.operators.sql import (
SQLCheckOperator,
SQLExecuteQueryOperator,
@@ -443,7 +443,7 @@ class SnowflakeSqlApiOperator(SQLExecuteQueryOperator):
if statement_status.get("status") == "success":
succeeded_query_ids.append(query_id)
else:
- raise AirflowException(f"{statement_status.get('status')}: {statement_status.get('message')}")
+ raise RuntimeError(f"{statement_status.get('status')}: {statement_status.get('message')}")
if len(self.query_ids) == len(succeeded_query_ids):
self.log.info("%s completed successfully.", self.task_id)
@@ -465,7 +465,7 @@ class SnowflakeSqlApiOperator(SQLExecuteQueryOperator):
while True:
statement_status = self.poll_on_queries()
if statement_status["error"]:
- raise AirflowException(statement_status["error"])
+ raise RuntimeError(str(statement_status["error"]))
if not statement_status["running"]:
break
@@ -509,7 +509,7 @@ class SnowflakeSqlApiOperator(SQLExecuteQueryOperator):
if event:
if "status" in event and event["status"] == "error":
msg = f"{event['status']}: {event['message']}"
- raise AirflowException(msg)
+ raise RuntimeError(msg)
if "status" in event and event["status"] == "success":
self.query_ids = cast("list[str]", event["statement_query_ids"])
self._hook.check_query_output(self.query_ids)
diff --git a/providers/snowflake/tests/unit/snowflake/hooks/test_snowflake_sql_api.py b/providers/snowflake/tests/unit/snowflake/hooks/test_snowflake_sql_api.py
index 062bee3a67f..7b2d397e8ab 100644
--- a/providers/snowflake/tests/unit/snowflake/hooks/test_snowflake_sql_api.py
+++ b/providers/snowflake/tests/unit/snowflake/hooks/test_snowflake_sql_api.py
@@ -35,7 +35,6 @@ from cryptography.hazmat.primitives.asymmetric import rsa
from airflow.exceptions import AirflowProviderDeprecationWarning
from airflow.models import Connection
-from airflow.providers.common.compat.sdk import AirflowException
from airflow.providers.snowflake.hooks.snowflake_sql_api import SnowflakeSqlApiHook
if TYPE_CHECKING:
@@ -391,14 +390,14 @@ class TestSnowflakeSqlApiHook:
mock_requests,
):
"""
- Test execute_query method by mocking the exception response and raise airflow exception
+ Test execute_query method by mocking the exception response and raise RuntimeError
without statementHandle in the response
"""
# status_code, json payload without statementHandle
mock_make_api_call.return_value = (None, {"foo": "bar"})
hook = SnowflakeSqlApiHook("mock_conn_id")
- with pytest.raises(AirflowException):
+ with pytest.raises(RuntimeError):
hook.execute_query(sql, statement_count)
@pytest.mark.parametrize(
@@ -517,11 +516,11 @@ class TestSnowflakeSqlApiHook:
@mock.patch(f"{HOOK_PATH}._get_conn_params")
def test_get_headers_pat_raises_when_password_missing(self, mock_conn_param):
- """Test get_headers raises AirflowException when PAT authenticator is set but password is empty."""
+ """Test get_headers raises ValueError when PAT authenticator is set but password is empty."""
conn_params_pat_no_password = {**CONN_PARAMS_PAT, "password": ""}
mock_conn_param.return_value = conn_params_pat_no_password
hook = SnowflakeSqlApiHook(snowflake_conn_id="mock_conn_id")
- with pytest.raises(AirflowException, match="Programmatic Access Token"):
+ with pytest.raises(ValueError, match="Programmatic Access Token"):
hook.get_headers()
@mock.patch("airflow.providers.snowflake.hooks.snowflake.HTTPBasicAuth")
@@ -629,7 +628,7 @@ class TestSnowflakeSqlApiHook:
"os.environ", AIRFLOW_CONN_TEST_CONN=Connection(**connection_kwargs).get_uri()
),
pytest.raises(
- AirflowException,
+ ValueError,
match="The private_key_file and private_key_content extra fields are mutually "
"exclusive. Please remove one.",
),
diff --git a/providers/snowflake/tests/unit/snowflake/operators/test_snowflake.py b/providers/snowflake/tests/unit/snowflake/operators/test_snowflake.py
index 400b813d64b..cca046f0415 100644
--- a/providers/snowflake/tests/unit/snowflake/operators/test_snowflake.py
+++ b/providers/snowflake/tests/unit/snowflake/operators/test_snowflake.py
@@ -27,7 +27,7 @@ from airflow.models import Connection
from airflow.models.dag import DAG
from airflow.models.dagrun import DagRun
from airflow.models.taskinstance import TaskInstance
-from airflow.providers.common.compat.sdk import AirflowException, TaskDeferred
+from airflow.providers.common.compat.sdk import TaskDeferred
from airflow.providers.common.sql.operators.sql import SQLExecuteQueryOperator
from airflow.providers.snowflake.operators.snowflake import (
SnowflakeCheckOperator,
@@ -339,7 +339,7 @@ class TestSnowflakeSqlApiOperator:
)
mock_execute_query.return_value = ["uuid1", "uuid2"]
mock_get_sql_api_query_status.side_effect = [{"status": "error"}, {"status": "success"}]
- with pytest.raises(AirflowException):
+ with pytest.raises(RuntimeError):
operator.execute(context=None)
@pytest.mark.parametrize(
@@ -373,7 +373,7 @@ class TestSnowflakeSqlApiOperator:
)
def test_snowflake_sql_api_execute_complete_failure(self):
- """Test SnowflakeSqlApiOperator raise AirflowException of error event"""
+ """Test SnowflakeSqlApiOperator raise RuntimeError of error event"""
operator = SnowflakeSqlApiOperator(
task_id=TASK_ID,
@@ -382,7 +382,7 @@ class TestSnowflakeSqlApiOperator:
statement_count=4,
deferrable=True,
)
- with pytest.raises(AirflowException):
+ with pytest.raises(RuntimeError):
operator.execute_complete(
context=None,
event={"status": "error", "message": "Test failure message", "type": "FAILED_WITH_ERROR"},
@@ -469,7 +469,7 @@ class TestSnowflakeSqlApiOperator:
)
mock_execute_query.return_value = ["uuid1"]
mock_get_sql_api_query_status.side_effect = [{"status": "error"}]
- with pytest.raises(AirflowException):
+ with pytest.raises(RuntimeError):
operator.execute(create_context(operator))
assert not mock_defer.called
@@ -551,7 +551,7 @@ class TestSnowflakeSqlApiOperator:
self, mock_execute_query, mock_get_sql_api_query_status, mock_check_query_output
):
"""
- Tests that the execute method raises AirflowException if any query fails during polling
+ Tests that the execute method raises RuntimeError if any query fails during polling
when ``deferrable=False``
"""
operator = SnowflakeSqlApiOperator(
@@ -572,7 +572,7 @@ class TestSnowflakeSqlApiOperator:
{"status": "error"},
]
- with pytest.raises(AirflowException):
+ with pytest.raises(RuntimeError):
operator.execute(context=None)
mock_check_query_output.assert_not_called()
diff --git a/scripts/ci/prek/known_airflow_exceptions.txt b/scripts/ci/prek/known_airflow_exceptions.txt
index 6175e6ad00c..b9dd36f80c2 100644
--- a/scripts/ci/prek/known_airflow_exceptions.txt
+++ b/scripts/ci/prek/known_airflow_exceptions.txt
@@ -392,10 +392,6 @@ providers/slack/src/airflow/providers/slack/transfers/sql_to_slack.py::1
providers/slack/src/airflow/providers/slack/transfers/sql_to_slack_webhook.py::3
providers/smtp/src/airflow/providers/smtp/hooks/smtp.py::16
providers/smtp/src/airflow/providers/smtp/operators/smtp.py::3
-providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake.py::1
-providers/snowflake/src/airflow/providers/snowflake/hooks/snowflake_sql_api.py::3
-providers/snowflake/src/airflow/providers/snowflake/operators/snowflake.py::3
-providers/snowflake/tests/unit/snowflake/operators/test_snowflake.py::1
providers/ssh/src/airflow/providers/ssh/hooks/ssh.py::5
providers/ssh/src/airflow/providers/ssh/operators/ssh.py::3
providers/ssh/src/airflow/providers/ssh/operators/ssh_remote_job.py::7