Repository: apache/airflow
Pull request: Add Snowflake support to SQL operator and sensor
Add Snowflake as supported database backend for SQL operator and sensor. This is primarily an update to the `allowed_conn_type` sets for the corresponding classes.

Closes #9548 
---
Make sure to mark the boxes below before creating PR: [x]

- [X] Description above provides context of the change
- [x] Unit tests coverage for changes (not needed for documentation changes)
- [X] Target Github ISSUE in description if exists
- [X] Commits follow "[How to write a good git commit message](http://chris.beams.io/posts/git-commit/)"
- [X] Relevant documentation is updated including usage instructions.
- [X] I will engage committers as explained in [Contribution Workflow Example](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#contribution-workflow-example).

---
In case of fundamental code change, Airflow Improvement Proposal ([AIP](https://cwiki.apache.org/confluence/display/AIRFLOW/Airflow+Improvements+Proposals)) is needed.
In case of a new dependency, check compliance with the [ASF 3rd Party License Policy](https://www.apache.org/legal/resolved.html#category-x).
In case of backwards incompatible changes please leave a note in [UPDATING.md](https://github.com/apache/airflow/blob/master/UPDATING.md).
Read the [Pull Request Guidelines](https://github.com/apache/airflow/blob/master/CONTRIBUTING.rst#pull-request-guidelines) for more information.


Problem statement:
Add support for snowflake on the sql_sensor
**Description**

Enable the support for Snowflake on the SqlSensor.

**Use case**

Currently, the `SqlSensor` supports only the following connection types,

- google_cloud_platform
- jdbc
- mssql
- mysql
- oracle
- postgres
- presto
- sqlite
- vertica

At the moment, it doesn't have support for Snowflake. This ticket is to add support for Snowflake on the `SqlSensor`.

Hints from issue discussion:
Thanks for opening your first issue here! Be sure to follow the issue template!

Hi, I would like to pick up this issue if it is a good feature to come in.
@KumarManoj-S I assigned you. to this ticket. This change seems sensible. We should not have lists of approved classes in the code, but we should rely on interfaces.
@KumarManoj-S  note that if this works for SqlSensor there is no reason it won't work for SqlOperators
https://github.com/apache/airflow/blob/1d36b0303b8632fce6de78ca4e782ae26ee06fea/airflow/sensors/sql_sensor.py#L72
https://github.com/apache/airflow/blob/0b9bf4a285a074bbde270839a90fb53c257340be/airflow/operators/sql.py#L26
 (not sure why two different lists are needed as lists are the same and there shouldn't be a reason why sensor would work on specific db but the operators won't or vice versa)
Might worth checking that if you can address both
@eladkal Sure. That's a good point. I will check if I can address both of them.

Commit message: Add test for conn_type to valid hook mapping

Patch to review:
diff --git a/airflow/models/connection.py b/airflow/models/connection.py
index 7144bcba79007..75174c41784cc 100644
--- a/airflow/models/connection.py
+++ b/airflow/models/connection.py
@@ -72,7 +72,7 @@
     "jira": ("airflow.providers.jira.hooks.jira.JiraHook", "jira_conn_id"),
     "kubernetes": ("airflow.providers.cncf.kubernetes.hooks.kubernetes.KubernetesHook", "kubernetes_conn_id"),
     "mongo": ("airflow.providers.mongo.hooks.mongo.MongoHook", "conn_id"),
-    "mssql": ("airflow.providers.microsoft.mssql.hooks.mssql.MsSqlHook", "mssql_conn_id"),
+    "mssql": ("airflow.providers.odbc.hooks.odbc.OdbcHook", "odbc_conn_id"),
     "mysql": ("airflow.providers.mysql.hooks.mysql.MySqlHook", "mysql_conn_id"),
     "odbc": ("airflow.providers.odbc.hooks.odbc.OdbcHook", "odbc_conn_id"),
     "oracle": ("airflow.providers.oracle.hooks.oracle.OracleHook", "oracle_conn_id"),
@@ -80,6 +80,7 @@
     "postgres": ("airflow.providers.postgres.hooks.postgres.PostgresHook", "postgres_conn_id"),
     "presto": ("airflow.providers.presto.hooks.presto.PrestoHook", "presto_conn_id"),
     "redis": ("airflow.providers.redis.hooks.redis.RedisHook", "redis_conn_id"),
+    "snowflake": ("airflow.providers.snowflake.hooks.snowflake.SnowflakeHook", "snowflake_conn_id"),
     "sqlite": ("airflow.providers.sqlite.hooks.sqlite.SqliteHook", "sqlite_conn_id"),
     "tableau": ("airflow.providers.salesforce.hooks.tableau.TableauHook", "tableau_conn_id"),
     "vertica": ("airflow.providers.vertica.hooks.vertica.VerticaHook", "vertica_conn_id"),
diff --git a/airflow/operators/sql.py b/airflow/operators/sql.py
index a0427f3eb959b..c99933f1e5acb 100644
--- a/airflow/operators/sql.py
+++ b/airflow/operators/sql.py
@@ -32,6 +32,7 @@
     "oracle",
     "postgres",
     "presto",
+    "snowflake",
     "sqlite",
     "vertica",
 }
diff --git a/airflow/sensors/sql_sensor.py b/airflow/sensors/sql_sensor.py
index e4f8e3aa6a7d5..4996ee5d416a3 100644
--- a/airflow/sensors/sql_sensor.py
+++ b/airflow/sensors/sql_sensor.py
@@ -71,7 +71,7 @@ def _get_hook(self):
 
         allowed_conn_type = {'google_cloud_platform', 'jdbc', 'mssql',
                              'mysql', 'odbc', 'oracle', 'postgres',
-                             'presto', 'sqlite', 'vertica'}
+                             'presto', 'snowflake', 'sqlite', 'vertica'}
         if conn.conn_type not in allowed_conn_type:
             raise AirflowException("The connection type is not supported by SqlSensor. " +
                                    "Supported connection types: {}".format(list(allowed_conn_type)))
diff --git a/tests/models/test_connection.py b/tests/models/test_connection.py
index 3b6abb27ce344..35663f0282212 100644
--- a/tests/models/test_connection.py
+++ b/tests/models/test_connection.py
@@ -30,6 +30,7 @@
 from airflow.models import Connection, crypto
 from airflow.models.connection import CONN_TYPE_TO_HOOK
 from airflow.providers.sqlite.hooks.sqlite import SqliteHook
+from airflow.utils.module_loading import import_string
 from tests.test_utils.config import conf_vars
 
 ConnectionParts = namedtuple("ConnectionParts", ["conn_type", "login", "password", "host", "port", "schema"])
@@ -533,3 +534,7 @@ def test_enforce_alphabetical_order(self):
         expected_keys = sorted(current_keys)
 
         self.assertEqual(expected_keys, current_keys)
+
+    def test_hooks_importable(self):
+        for conn_type in CONN_TYPE_TO_HOOK:
+            self.assertTrue(issubclass(import_string(CONN_TYPE_TO_HOOK[conn_type][0]), BaseHook))


Code review comments:
diff_hunk: @@ -533,3 +534,7 @@ def test_enforce_alphabetical_order(self):
         expected_keys = sorted(current_keys)
 
         self.assertEqual(expected_keys, current_keys)
+
+    def test_hooks_importable(self):
+        for conn_type in CONN_TYPE_TO_HOOK:
+            self.assertTrue(issubclass(import_string(CONN_TYPE_TO_HOOK[conn_type][0]), BaseHook))
path: tests/models/test_connection.py
text: @user1:
```suggestion
        for hook_path, _ in CONN_TYPE_TO_HOOK.values():
            self.assertTrue(issubclass(import_string(hook_path), BaseHook))
```
WDYT?

@author:
LGTM. This is much cleaner.