Fix Snowflake hook conn id (#8423)

Do not set manually to ensure the base `DbApiHook` correctly interprets
arguments based on `conn_name_attr` and `default_conn_name`.

UPDATING.md

@@ -1133,6 +1133,10 @@ If the DAG relies on tasks with other trigger rules (i.e. `all_done`) being skip
 
 The goal of this change is to achieve a more consistent and configurale cascading behaviour based on the `BaseBranchOperator` (see [AIRFLOW-2923](https://jira.apache.org/jira/browse/AIRFLOW-2923) and [AIRFLOW-1784](https://jira.apache.org/jira/browse/AIRFLOW-1784)).
 
+### Change default snowflake_conn_id for Snowflake hook and operators
+
+When initializing a Snowflake hook or operator, the value used for `snowflake_conn_id` was always `snowflake_conn_id`, regardless of whether or not you specified a value for it. The default `snowflake_conn_id` value is now switched to `snowflake_default` for consistency and will be properly overriden when specified.
+
 ## Airflow 1.10.10
 
 ### Setting Empty string to a Airflow Variable will return an empty string

airflow/providers/snowflake/hooks/snowflake.py

@@ -41,14 +41,13 @@ def __init__(self, *args, **kwargs):
         self.region = kwargs.pop("region", None)
         self.role = kwargs.pop("role", None)
         self.schema = kwargs.pop("schema", None)
-        self.snowflake_conn_id = 'snowflake_conn_id'
 
     def _get_conn_params(self):
         """
         one method to fetch connection params as a dict
         used in get_uri() and get_connection()
         """
-        conn = self.get_connection(self.snowflake_conn_id)
+        conn = self.get_connection(self.snowflake_conn_id)  # pylint: disable=no-member
         account = conn.extra_dejson.get('account', '')
         warehouse = conn.extra_dejson.get('warehouse', '')
         database = conn.extra_dejson.get('database', '')
@@ -119,8 +118,8 @@ def _get_aws_credentials(self):
 
         intended to be used by external import and export statements
         """
-        if self.snowflake_conn_id:
-            connection_object = self.get_connection(self.snowflake_conn_id)
+        if self.snowflake_conn_id:  # pylint: disable=no-member
+            connection_object = self.get_connection(self.snowflake_conn_id)  # pylint: disable=no-member
             if 'aws_secret_access_key' in connection_object.extra_dejson:
                 aws_access_key_id = connection_object.extra_dejson.get(
                     'aws_access_key_id')

tests/providers/snowflake/hooks/test_snowflake.py

@@ -102,6 +102,7 @@ def test_get_conn_params(self):
                                 'warehouse': 'af_wh',
                                 'region': 'af_region',
                                 'role': 'af_role'}
+        self.assertEqual(self.db_hook.snowflake_conn_id, 'snowflake_default')  # pylint: disable=no-member
         self.assertEqual(conn_params_shouldbe, self.db_hook._get_conn_params())
 
     def test_get_conn(self):