Connecting to Snowflake with SQLAlchemy¶

Connection parameters¶

Snowflake SQLAlchemy uses the following syntax for the connection string used to connect to Snowflake and initiate a session:

'snowflake://<user_login_name>:<password>@<account_identifier>'

Where:

• <user_login_name> is the login name for your Snowflake user.
• <password> is the password for your Snowflake user.
• <account_identifier> is the identifier for your Snowflake account.

Include the region in the <account_identifier> if applicable. Don’t include the snowflakecomputing.com domain. For more information, see Account name in your organization.

You can optionally specify the initial database and schema for the Snowflake session by including them at the end of the connection string, separated by /. You can also specify the initial warehouse and role for the session as a parameter string at the end of the connection string:

'snowflake://<user_login_name>:<password>@<account_identifier>/<database_name>/<schema_name>?warehouse=<warehouse_name>&role=<role_name>'

Escaping special characters in passwords¶

As noted in SQLAlchemy documentation, URLs containing special characters need to be URL encoded to be parsed correctly. This includes the % and @ signs. Unescaped passwords containing special characters could lead to authentication failure.

The encoding for the password can be generated using urllib.parse:

import urllib.parse
urllib.parse.quote("kx@% jj5/g")
'kx%40%25%20jj5/g'

To create an engine with the proper encodings, either manually construct the URL string by formatting or use the snowflake.sqlalchemy.URL helper method:

import urllib.parse
from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine

quoted_password = urllib.parse.quote("kx@% jj5/g")

# 1. Manually constructing a URL string
url = f'snowflake://testuser1:{quoted_password}@abc123/testdb/public?warehouse=testwh&role=myrole'
engine = create_engine(url)

# 2. Using the snowflake.sqlalchemy.URL helper method
engine = create_engine(URL(
    account='abc123',
    user='testuser1',
    password=quoted_password,
    database='testdb',
    schema='public',
    warehouse='testwh',
    role='myrole',
))

The URL helper validates the account and region values, which may contain only letters, digits, hyphens, dots, and underscores, and raises sqlalchemy.exc.ArgumentError otherwise. It also percent-encodes user, so characters such as ?, &, @, and # can’t inject extra URL parameters. URL doesn’t accept private key parameters; pass those through connect_args (see Key pair authentication support).

The following example calls the create_engine method with the user name testuser1, password 0123456, account name abc123, database testdb, schema public, warehouse testwh, and role myrole:

from sqlalchemy import create_engine
engine = create_engine(
    'snowflake://testuser1:0123456@abc123/testdb/public?warehouse=testwh&role=myrole'
)

Other connector parameters, such as timezone, can be passed as a URL argument or through connect_args:

from sqlalchemy import create_engine

engine = create_engine(
    'snowflake://testuser1:0123456@abc123/testdb/public?warehouse=testwh&role=myrole',
    connect_args={
        'timezone': 'America/Los_Angeles',
    }
)

To fetch NumPy data types (numpy.int64, numpy.float64, and numpy.datetime64), add 'numpy': True to connect_args. Binding NumPy values works without this setting.

Using the create_snowflake_engine helper¶

To connect with a case-sensitive (mixed-case) schema name, use create_snowflake_engine() instead of building the URL by hand. It quotes the schema name for you. Other keyword arguments are passed to create_engine():

from snowflake.sqlalchemy import create_snowflake_engine

engine = create_snowflake_engine(
    'snowflake://user:password@account/database',
    schema='MixedCaseSchema',
    case_sensitive_schema=True,
)

The base URL must not already include a schema. For the async equivalent, see Async support in Snowflake SQLAlchemy.

Using a proxy server¶

The Snowflake Connector for Python supports tunneling connections through a proxy server. To use a proxy server with Snowflake SQLAlchemy, pass the proxy parameters through connect_args. You can also set proxy_scheme to http (the default) or https:

from sqlalchemy import create_engine

engine = create_engine(
    'snowflake://user:password@account/database/schema',
    connect_args={
        'proxy_host': 'proxy.example.com',
        'proxy_port': 8080,
        'proxy_user': 'proxy_user',
        'proxy_password': 'proxy_password',
    }
)

When no explicit proxy parameters are set, the connector uses the HTTP_PROXY, HTTPS_PROXY, and NO_PROXY environment variables. To ignore them, set 'use_proxy_env': False in connect_args.

Using session parameters¶

Session parameters specify the default behavior for the session. Pass them through the session_parameters dictionary in connect_args:

from sqlalchemy import create_engine

engine = create_engine(
    'snowflake://user:password@account/database/schema',
    connect_args={
        'session_parameters': {
            'STATEMENT_TIMEOUT_IN_SECONDS': 10,
            'QUERY_TAG': 'my_tag',
            'QUOTED_IDENTIFIERS_IGNORE_CASE': True,
        }
    }
)

Session parameters set this way apply to every query in the session. To change a parameter for specific queries, run ALTER SESSION:

from sqlalchemy import text

with engine.connect() as conn:
    conn.execute(text("ALTER SESSION SET QUERY_TAG = 'batch_job_1'"))
    conn.execute(text("..."))  # Uses 'batch_job_1'

    conn.execute(text("ALTER SESSION UNSET QUERY_TAG"))
    conn.execute(text("..."))  # No tag

Sensitive connection parameters¶

A set of connector parameters is blocked from the URL query string because they change the connection target, read or write local files, or relax a connector safety check. Passing any of them in the URL query string raises sqlalchemy.exc.ArgumentError. Pass them through connect_args in create_engine() instead:

• host
• protocol
• token_file_path
• private_key_file
• ocsp_response_cache_filename
• connection_diag_log_path
• crl_cache_dir
• unsafe_file_write
• unsafe_skip_file_permissions_check
• unsafe_skip_config_file_permissions_check
• crl_unsafe_skip_file_permissions_check

For example:

from sqlalchemy import create_engine

engine = create_engine(
    'snowflake://user:password@account/',
    connect_args={
        'protocol': 'https',
        'token_file_path': '/path/to/token',
    }
)

Dialect options¶

Pass these options as keyword arguments to create_engine():

The legacy_url_params and force_div_is_floordiv options were removed. Passing either to create_engine() raises an error. The / operator always performs true division, and // always performs floor division.

Strict host-name mode¶

For air-gapped or custom-domain deployments, set strict_host_names=True in create_engine() (or set the SNOWFLAKE_SQLALCHEMY_STRICT_HOST_NAMES=true environment variable). In this mode the dialect doesn’t infer or append any domain to host names. It forwards connection parameters, such as host, port, authentication, and connection_name, to the Snowflake Connector for Python unchanged and doesn’t validate them; the connector handles connection establishment, including missing parameters. When strict mode is off (the default), behavior is unchanged.

Supply all connection parameters through connect_args:

from sqlalchemy import create_engine

engine = create_engine(
    'snowflake://',
    strict_host_names=True,
    connect_args={
        'account': 'myaccount',
        'host': 'myaccount.snowflake.example.com',
        'port': 443,
        'user': 'myuser',
        'authenticator': 'SNOWFLAKE_JWT',
        'private_key_file': '/path/to/rsa_key.p8',
        'database': 'mydb',
        'schema': 'public',
        'warehouse': 'mywh',
    },
)

Or use a named profile from the connector’s connections.toml, which the connector reads and applies itself:

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL

engine = create_engine(
    URL(account=''),
    strict_host_names=True,
    connect_args={'connection_name': 'my_profile'},
)

AzureContainer accepts an explicit endpoint (for example, endpoint='blob.storage.example.com'). In strict mode, an AzureContainer without one fails at SQL compilation instead of defaulting to .blob.core.windows.net.

Opening and closing a connection¶

To open a connection:

connection = engine.connect()

To close the connection:

connection.close()

To dispose of the engine:

engine.dispose()

Prefer a context manager, which closes the connection for you:

from sqlalchemy import text

with engine.connect() as connection:
    connection.execute(text('select 1'))

When using SQLAlchemy sessions, connections are managed automatically. Typically, you don’t need to call close() or dispose() directly.

Connections use the READ COMMITTED isolation level with the connector’s autocommit turned off, so changes aren’t saved until you commit. Use connection.commit(), engine.begin(), or an ORM Session transaction.

Automatic reconnection on expired sessions¶

The dialect flags recoverable Snowflake session and token loss errors, such as “Authentication token has expired” or “Session no longer exists,” as disconnects. SQLAlchemy’s connection pool then discards the dead connection and opens a new one. To validate pooled connections before use, set pool_pre_ping=True:

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL

engine = create_engine(
    URL(account='myaccount', user='me', password='secret'),
    pool_pre_ping=True,
)

Permanent authentication failures, such as revoked credentials, aren’t treated as disconnects. They’re raised as errors, because reconnecting can’t fix them.