Issue 1:
Apache Airflow Provider(s)

snowflake

Versions of Apache Airflow Providers

apache-airflow-providers-snowflake | 2.4.0

Apache Airflow version

2.2.3 (latest released)

Operating System

Ubuntu 20.04

Deployment

Virtualenv installation

Deployment details

Python 3.8.10

alembic==1.7.5

antiorm==1.2.1

anyio==3.5.0

apache-airflow==2.2.3

apache-airflow-providers-ftp==2.0.1

apache-airflow-providers-http==2.0.2

apache-airflow-providers-imap==2.1.0

apache-airflow-providers-snowflake==2.4.0

apache-airflow-providers-sqlite==2.0.1

apispec==3.3.2

argcomplete==1.12.3

asn1crypto==1.4.0

attrs==20.3.0

Babel==2.9.1

blinker==1.4

cattrs==1.5.0

certifi==2021.10.8

cffi==1.15.0

charset-normalizer==2.0.10

click==7.1.2

clickclick==20.10.2

colorama==0.4.4

colorlog==5.0.1

commonmark==0.9.1

croniter==1.0.15

cryptography==3.4.8

db==0.1.1

db-sqlite3==0.0.1

defusedxml==0.7.1

dill==0.3.4

dnspython==2.2.0

docutils==0.16

email-validator==1.1.3

Flask==1.1.4

Flask-AppBuilder==3.4.3

Flask-Babel==2.0.0

Flask-Caching==1.10.1

Flask-JWT-Extended==3.25.1

Flask-Login==0.4.1

Flask-OpenID==1.3.0

Flask-SQLAlchemy==2.5.1

Flask-WTF==0.14.3

graphviz==0.19.1

gunicorn==20.1.0

h11==0.12.0

httpcore==0.13.7

httpx==0.19.0

idna==3.3

importlib-metadata==4.10.1

importlib-resources==5.4.0

inflection==0.5.1

iso8601==1.0.2

isodate==0.6.1

itsdangerous==1.1.0

Jinja2==2.11.3

jsonschema==3.2.0

lazy-object-proxy==1.7.1

lockfile==0.12.2

Mako==1.1.6

Markdown==3.3.6

MarkupSafe==2.0.1

marshmallow==3.14.1

marshmallow-enum==1.5.1

marshmallow-oneofschema==3.0.1

marshmallow-sqlalchemy==0.26.1

openapi-schema-validator==0.1.6

openapi-spec-validator==0.3.2

oscrypto==1.2.1

packaging==21.3

pendulum==2.1.2

pkg_resources==0.0.0

prison==0.2.1

psutil==5.9.0

pycparser==2.21

pycryptodomex==3.12.0

Pygments==2.11.2

PyJWT==1.7.1

pyOpenSSL==20.0.1

pyparsing==3.0.7

pyrsistent==0.16.1

python-daemon==2.3.0

python-dateutil==2.8.2

python-nvd3==0.15.0

python-slugify==4.0.1

python3-openid==3.2.0

pytz==2021.3

pytzdata==2020.1

PyYAML==6.0

requests==2.27.1

rfc3986==1.5.0

rich==11.0.0

setproctitle==1.2.2

six==1.16.0

sniffio==1.2.0

snowflake-connector-python==2.7.1

snowflake-sqlalchemy==1.2.4

SQLAlchemy==1.3.24

SQLAlchemy-JSONField==1.0.0

SQLAlchemy-Utils==0.38.2

swagger-ui-bundle==0.0.9

tabulate==0.8.9

tenacity==8.0.1

termcolor==1.1.0

text-unidecode==1.3

unicodecsv==0.14.1

urllib3==1.26.8

Werkzeug==1.0.1

WTForms==2.3.3

zipp==3.7.0

What happened

After creating a new Snowflake connection in the Airflow web portal, the Extras parameter is automatically set to the boolean value false (see example below).

{"extra__snowflake__account": "******", "extra__snowflake__database": "", "extra__snowflake__insecure_mode": false, "extra__snowflake__region": "canada-central.azure", "extra__snowflake__role": "", "extra__snowflake__warehouse": ""}

When running or testing a dag, an error is thrown in SnowflakeHook._get_conn_params when calling airflow.utils.strings.to_boolean on the bool value while evaluating insecure_mode. See end of stack trace below.

conn_config = self._get_conn_params()

File "/home/gdorval/airflow/lib/python3.8/site-packages/airflow/providers/snowflake/hooks/snowflake.py", line 170, in _get_conn_params

insecure_mode = to_boolean(

File "/home/gdorval/airflow/lib/python3.8/site-packages/airflow/utils/strings.py", line 30, in to_boolean

return False if astring is None else astring.lower() in ['true', 't', 'y', 'yes', '1']

AttributeError: 'bool' object has no attribute 'lower'

What you expected to happen

SnowflakeHook._get_conn_params should expect a boolean value for "extra__snowflake__insecure_mode". There is no need to convert it from a string to boolean.

How to reproduce

Create a new Snowflake connection.

Create a simple dag with one task using the snowflake connection (example below).

Test the task

query_test = """select current_date() as date;"""

with DAG(
'example_snowflake',
start_date=datetime(2021, 1, 1),
default_args=DEFAULT_ARGS,
tags=['snowflake'],
catchup=False,
) as dag:

snowflake_op_sql_str = SnowflakeOperator(
task_id='snowflake_op_sql_str',
dag=dag,
sql=query_test,
snowflake_conn_id=SNOWFLAKE_CONN_ID,
warehouse=SNOWFLAKE_WAREHOUSE,
database=SNOWFLAKE_DATABASE,
schema=SNOWFLAKE_SCHEMA,
role=SNOWFLAKE_ROLE,
)

snowflake_op_sql_str

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct

Issue 2:
Apache Airflow version

2.2.3 (latest released)

What happened

Error thrown when using SnowflakeOperator with a Snowflake Connection.

After creating a Snowflake Connection, the "Extra" field was automatically filled with a dictionary containing the values entered in the other input fields. Note: the value for the key "extra__snowflake__insecure_mode" is a boolean.

A task using SnowflakeOperator fails, throwing following error:

[2022-01-25, 14:39:54 UTC] {taskinstance.py:1700} ERROR - Task failed with exception
Traceback (most recent call last):
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1329, in _run_raw_task
self._execute_task_with_callbacks(context)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1455, in _execute_task_with_callbacks
result = self._execute_task(context, self.task)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1511, in _execute_task
result = execute_callable(context=context)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/providers/snowflake/operators/snowflake.py", line 129, in execute
execution_info = hook.run(self.sql, autocommit=self.autocommit, parameters=self.parameters)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/providers/snowflake/hooks/snowflake.py", line 293, in run
with closing(self.get_conn()) as conn:
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/providers/snowflake/hooks/snowflake.py", line 236, in get_conn
conn_config = self._get_conn_params()
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/providers/snowflake/hooks/snowflake.py", line 170, in _get_conn_params
insecure_mode = to_boolean(
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/utils/strings.py", line 30, in to_boolean
return False if astring is None else astring.lower() in ['true', 't', 'y', 'yes', '1']
AttributeError: 'bool' object has no attribute 'lower'
[2022-01-25, 14:39:54 UTC] {taskinstance.py:1267} INFO - Marking task as FAILED. dag_id=test, task_id=snowflake_task, execution_date=20220123T000000, start_date=20220125T133954, end_date=20220125T133954
[2022-01-25, 14:39:55 UTC] {standard_task_runner.py:89} ERROR - Failed to execute job 4 for task snowflake_task
Traceback (most recent call last):
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/task/task_runner/standard_task_runner.py", line 85, in _start_by_fork
args.func(args, dag=self.dag)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/cli/cli_parser.py", line 48, in command
return func(*args, **kwargs)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/utils/cli.py", line 92, in wrapper
return f(*args, **kwargs)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 298, in task_run
_run_task_by_selected_method(args, dag, ti)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 107, in _run_task_by_selected_method
_run_raw_task(args, ti)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/cli/commands/task_command.py", line 180, in _run_raw_task
ti._run_raw_task(
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/utils/session.py", line 70, in wrapper
return func(*args, session=session, **kwargs)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1329, in _run_raw_task
self._execute_task_with_callbacks(context)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1455, in _execute_task_with_callbacks
result = self._execute_task(context, self.task)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/models/taskinstance.py", line 1511, in _execute_task
result = execute_callable(context=context)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/providers/snowflake/operators/snowflake.py", line 129, in execute
execution_info = hook.run(self.sql, autocommit=self.autocommit, parameters=self.parameters)
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/providers/snowflake/hooks/snowflake.py", line 293, in run
with closing(self.get_conn()) as conn:
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/providers/snowflake/hooks/snowflake.py", line 236, in get_conn
conn_config = self._get_conn_params()
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/providers/snowflake/hooks/snowflake.py", line 170, in _get_conn_params
insecure_mode = to_boolean(
File "/home/test_airflow2/.local/lib/python3.8/site-packages/airflow/utils/strings.py", line 30, in to_boolean
return False if astring is None else astring.lower() in ['true', 't', 'y', 'yes', '1']
AttributeError: 'bool' object has no attribute 'lower'
[2022-01-25, 14:39:55 UTC] {local_task_job.py:154} INFO - Task exited with return code 1
[2022-01-25, 14:39:55 UTC] {local_task_job.py:264} INFO - 0 downstream tasks scheduled from follow-on schedule check

This error seems to arise because the boolean value of extra__snowflake__insecure_mode is interpreted as a string.

Manually modifying the boolean extra__snowflake__insecure_mode value to be string in the "Extra" dictionary solves this problem:

false->"false"

What you expected to happen

Be able to create a usable Snowflake Connection, by filling in fields other than "Extra". The "Extra" field should be automatically filled with a correct/usable connection dictionary.

I can then use this Snowflake Connection for SnowflakeOperators in DAGs.

How to reproduce

Create a new Connection of type Snowflake, set an arbitrary Connection ID. The rest of the fields can be left empty (doesn't affect error).

Create a DAG with a SnowflakeOperator task, which uses the created Snowflake Connection:

from airflow import DAG
from airflow.providers.snowflake.operators.snowflake import SnowflakeOperator
from airflow.utils.dates import days_ago

with DAG('test', start_date=days_ago(2)) as dag:

snowflake_task = SnowflakeOperator(task_id='snowflake_task',
sql='select 1;',
snowflake_conn_id='snowflake_conn')

Execute the DAG, the task will fail and throw the above error.

Operating System

Ubuntu 20.04.2 LTS

Versions of Apache Airflow Providers

apache-airflow-providers-snowflake==2.4.0

Deployment

Virtualenv installation

Deployment details

No response

Anything else

No response

Are you willing to submit PR?

Yes I am willing to submit a PR!

Code of Conduct

I agree to follow this project's Code of Conduct