Snowflake SQLAlchemy runs on the top of the Snowflake Connector for Python as a dialect to bridge a Snowflake database and SQLAlchemy applications.

v1.11.0 — Sensitive connection parameters: We curated a set of connector parameters (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) that can no longer be supplied through the URL query string — pass them via connect_args in create_engine() instead. If you encounter a possible behavioral change, set SNOWFLAKE_SQLALCHEMY_LEGACY_URL_PARAMS=1 and follow the instructions in Sensitive connection parameters.

SQLAlchemy version 1.4 is a legacy version (see reference), which is why we are working on a version that supports only SQLAlchemy 2.x. The version will be available as a "release candidate" to enable earlier testing. We will publish more details together with its release. The current version will still be supported for two years in accordance with https://docs.snowflake.com/en/release-notes/requirements#recommended-client-versions

Table of contents:

Snowflake SQLAlchemy

Prerequisites

Snowflake Connector for Python

The only requirement for Snowflake SQLAlchemy is the Snowflake Connector for Python; however, the connector does not need to be installed because installing Snowflake SQLAlchemy automatically installs the connector.

Data Analytics and Web Application Frameworks (Optional)

Snowflake SQLAlchemy can be used with Pandas, Jupyter and Pyramid, which provide higher levels of application frameworks for data analytics and web applications. However, building a working environment from scratch is not a trivial task, particularly for novice users. Installing the frameworks requires C compilers and tools, and choosing the right tools and versions is a hurdle that might deter users from using Python applications.

An easier way to build an environment is through Anaconda, which provides a complete, precompiled technology stack for all users, including non-Python experts such as data analysts and students. For Anaconda installation instructions, see the Anaconda install documentation. The Snowflake SQLAlchemy package can then be installed on top of Anaconda using pip.

Installing Snowflake SQLAlchemy

The Snowflake SQLAlchemy package can be installed from the public PyPI repository using pip:

pip install --upgrade snowflake-sqlalchemy

pip automatically installs all required modules, including the Snowflake Connector for Python.

Verifying Your Installation

Create a file (e.g. validate.py) that contains the following Python sample code, which connects to Snowflake and displays the Snowflake version:

from sqlalchemy import create_engine
engine = create_engine(
    'snowflake://{user}:{password}@{account}/'.format(
        user='<your_user_login_name>',
        password='<your_password>',
        account='<your_account_name>',
    )
)
try:
    connection = engine.connect()
    results = connection.execute('select current_version()').fetchone()
    print(results[0])
finally:
    connection.close()
    engine.dispose()

Replace <your_user_login_name>, <your_password>, and <your_account_name> with the appropriate values for your Snowflake account and user.

For more details, see Connection Parameters.
Execute the sample code. For example, if you created a file named validate.py:
```
python validate.py
```
The Snowflake version (e.g. 1.48.0) should be displayed.

Parameters and Behavior

As much as possible, Snowflake SQLAlchemy provides compatible functionality for SQLAlchemy applications. For information on using SQLAlchemy, see the SQLAlchemy documentation.

However, Snowflake SQLAlchemy also provides Snowflake-specific parameters and behavior, which are described in the following sections.

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_name>'

Where:

<user_login_name> is the login name for your Snowflake user.
<password> is the password for your Snowflake user.
<account_name> is the name of your Snowflake account.

Include the region in the <account_name> if applicable, more info is available here.

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_name>/<database_name>/<schema_name>?warehouse=<warehouse_name>&role=<role_name>'

Escaping Special Characters such as `%, @` signs in Passwords

As pointed out in SQLAlchemy, URLs containing special characters need to be URL encoded to be parsed correctly. This includes the %, @ signs. Unescaped password 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'

Note: urllib.parse.quote_plus may also be used if there is no space in the string, as urllib.parse.quote_plus will replace space with +.

To create an engine with the proper encodings, either manually constructing the url string by formatting or taking advantage of 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 an 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',
))

Note: After login, the initial database, schema, warehouse and role specified in the connection string can always be changed for the session.

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 parameters, such as timezone, can also be specified as a URI parameter or in connect_args parameters. For example:

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

For convenience, you can use the snowflake.sqlalchemy.URL method to construct the connection string and connect to the database. The following example constructs the same connection string from the previous example:

from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine
engine = create_engine(URL(
    account = 'abc123',
    user = 'testuser1',
    password = '0123456',
    database = 'testdb',
    schema = 'public',
    warehouse = 'testwh',
    role='myrole',
    timezone = 'America/Los_Angeles',
))

Using a proxy server

Use the supported environment variables, HTTPS_PROXY, HTTP_PROXY and NO_PROXY to configure a proxy server.

Using session parameters

Snowflake session parameters (such as QUERY_TAG) cannot be set directly through the URL helper. Instead, pass them via the connect_args parameter of create_engine, using the session_parameters dict — the same way you would through the Python connector:

from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine
engine = create_engine(
    URL(
        # CONNECTION_PARAMETERS
    ),
    connect_args={
        "session_parameters": {
            "QUERY_TAG": "SOME_QUERY_TAGS",
        }
    },
)

Session parameters set this way apply to all queries executed within the session. To change a session parameter for specific queries mid-session, use 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 SET QUERY_TAG = 'batch_job_2'"))
    conn.execute(text("..."))  # Uses 'batch_job_2'
    conn.execute(text("ALTER SESSION UNSET QUERY_TAG"))
    conn.execute(text("..."))  # No tag

Sensitive connection parameters

For more robust connection-string handling, a small set of sensitive connector parameters cannot be supplied through the URL query string. These are parameters that change the connection target or transport, read or write local files, or relax a connector safety check:

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.

Passing any of these in the URL query string raises sqlalchemy.exc.ArgumentError. Supply them through the connect_args parameter of create_engine instead, which is fully under your application's control:

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL
# Instead of '...?protocol=https&host=...', pass them via connect_args:
engine = create_engine(
    URL(
        account="abc123",
        user="testuser1",
        password="0123456",
        database="testdb",
        schema="public",
    ),
    connect_args={
        "protocol": "https",
    },
)

Similarly, the URL helper validates the account and region values (which become part of the connection URL authority) and percent-encodes user, so URL metacharacters such as ?, &, @, and # cannot introduce additional parameters.

Temporary compatibility: `legacy_url_params`

If an existing application still relies on the previous behavior, you can re-enable it during migration. When enabled, the parameters above are forwarded from the URL query string as before and a DeprecationWarning is emitted for each one. This shim is temporary and will be removed in a future release.

Enable it in either of two ways (an explicit kwarg always wins; the environment variable is a global fallback):

# Per-engine, under application control (recommended over the env variable):
engine = create_engine(url, legacy_url_params=True)

# Global fallback for the whole process:
export SNOWFLAKE_SQLALCHEMY_LEGACY_URL_PARAMS=1

legacy_url_params is accepted only as an engine kwarg or the environment variable — it is intentionally not read from the URL query string, so a caller who controls only the URL cannot use it to relax this handling. Note that the URL() helper's account / region validation runs before any engine exists, so only the environment variable relaxes it at URL-construction time.

Opening and Closing Connection

Open a connection by executing engine.connect(); avoid using engine.execute(). Make certain to close the connection by executing connection.close() before engine.dispose(); otherwise, the Python Garbage collector removes the resources required to communicate with Snowflake, preventing the Python connector from closing the session properly.

# Avoid this.
engine = create_engine(...)
engine.execute(<SQL>)
engine.dispose()
# Better.
engine = create_engine(...)
connection = engine.connect()
try:
  connection.execute(text(<SQL>))
finally:
    connection.close()
    engine.dispose()
# Best
try:
    with engine.connect() as connection:
        connection.execute(text(<SQL>))
        # or
        connection.exec_driver_sql(<SQL>)
finally:
    engine.dispose()

Automatic reconnection on expired sessions

The dialect implements is_disconnect, flagging Snowflake session/token-loss errors (for example "Authentication token has expired" or "Session no longer exists") as disconnects based on the connector's structured error codes. This lets SQLAlchemy's connection pool transparently discard the dead connection and open a fresh one, so long-lived engines recover instead of raising. Combine it with pool_pre_ping=True to validate connections before use:

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL
engine = create_engine(
    URL(account="myaccount", user="me", password="secret"),
    pool_pre_ping=True,  # invalidate + reconnect stale/expired connections automatically
)

Note: only recoverable session/token-loss errors are treated as disconnects. Permanent authentication failures (for example revoked credentials) are not flagged as disconnects, because reconnecting cannot recover them — they surface as errors so the underlying problem stays visible.

Auto-increment Behavior

Auto-incrementing a value requires the Sequence object. Include the Sequence object in the primary key column to automatically increment the value as each new record is inserted. For example:

t = Table('mytable', metadata,
    Column('id', Integer, Sequence('id_seq'), primary_key=True),
    Column(...), ...
)

Sequence is the recommended approach for ORM primary keys because the generated value is available to the ORM after insert. See Identity columns as primary keys in Known Limitations for why Identity() does not work as an ORM primary key.

AUTOINCREMENT columns. The dialect renders a column as AUTOINCREMENT when it is the table's autoincrement column and has no other server default. This is driven by SQLAlchemy's standard autoincrement=True:

t = Table('mytable', metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),  # rendered as AUTOINCREMENT
    Column('name', String),
)

Sequences and Hybrid tables. For Hybrid tables (which require a defined primary key), both Sequence-backed and AUTOINCREMENT columns work. Note the difference in how the key is produced (MyModel below is an ORM-mapped class over such a table):

With a Sequence, SQLAlchemy fetches the next value with a separate SELECT <seq>.nextval before the INSERT — on both the ORM and Core paths:

from sqlalchemy import insert
# ORM — issues SELECT id_seq.nextval, then INSERT
session.add(MyModel(name="abc"))
session.commit()
# Core — same two-step behavior
connection.execute(insert(MyModel).values(name="abc"))

With AUTOINCREMENT (no Sequence), Snowflake assigns the value server-side, so a plain insert is a single statement:
```
connection.execute(insert(MyModel).values(name="abc"))  # single INSERT; id assigned by Snowflake
```

Object Name Case Handling

Snowflake stores all case-insensitive object names in uppercase text. In contrast, SQLAlchemy considers all lowercase object names to be case-insensitive. Snowflake SQLAlchemy converts the object name case during schema-level communication, i.e. during table and index reflection. If you use uppercase object names, SQLAlchemy assumes they are case-sensitive and encloses the names with quotes. This behavior will cause mismatches against data dictionary data received from Snowflake, so unless identifier names have been truly created as case sensitive using quotes, e.g., "TestDb", all lowercase names should be used on the SQLAlchemy side.

Index Support

Indexes are supported only for Hybrid Tables in Snowflake SQLAlchemy. For more details on limitations and use cases, refer to the Create Index documentation. You can create an index using the following methods:

Single Column Index

You can create a single column index by setting the index=True parameter on the column or by explicitly defining an Index object.

hybrid_test_table_1 = HybridTable(
  "table_name",
  metadata,
  Column("column1", Integer, primary_key=True),
  Column("column2", String, index=True),
  Index("index_1", "column1", "column2")
)
metadata.create_all(engine_testaccount)

Multi-Column Index

For multi-column indexes, you define the Index object specifying the columns that should be indexed.

hybrid_test_table_1 = HybridTable(
  "table_name",
  metadata,
  Column("column1", Integer, primary_key=True),
  Column("column2", String),
  Index("index_1", "column1", "column2")
)
metadata.create_all(engine_testaccount)

Numpy Data Type Support

Snowflake SQLAlchemy supports binding and fetching NumPy data types. Binding is always supported. To enable fetching NumPy data types, add numpy=True to the connection parameters.

The following example shows the round trip of numpy.datetime64 data:

import numpy as np
import pandas as pd
engine = create_engine(URL(
    account = 'abc123',
    user = 'testuser1',
    password = 'pass',
    database = 'db',
    schema = 'public',
    warehouse = 'testwh',
    role='myrole',
    numpy=True,
))
specific_date = np.datetime64('2016-03-04T12:03:05.123456789Z')
with engine.connect() as connection:
    connection.exec_driver_sql(
        "CREATE OR REPLACE TABLE ts_tbl(c1 TIMESTAMP_NTZ)")
    connection.exec_driver_sql(
        "INSERT INTO ts_tbl(c1) values(%s)", (specific_date,)
    )
    df = pd.read_sql_query("SELECT * FROM ts_tbl", connection)
    assert df.c1.values[0] == specific_date

The following NumPy data types are supported:

numpy.int64
numpy.float64
numpy.datatime64

DECFLOAT Data Type Support

Snowflake SQLAlchemy supports the DECFLOAT data type, which provides decimal floating-point with up to 38 significant digits. For more information, see the Snowflake DECFLOAT documentation.

from sqlalchemy import Column, Integer, MetaData, Table
from snowflake.sqlalchemy import DECFLOAT
metadata = MetaData()
t = Table('my_table', metadata,
    Column('id', Integer, primary_key=True),
    Column('value', DECFLOAT()),
)
metadata.create_all(engine)

DECFLOAT Precision

The Snowflake Python connector uses Python's decimal module context when converting DECFLOAT values to Python Decimal objects. Python's default decimal context precision is 28 digits, which can truncate DECFLOAT values that use up to 38 digits.

To preserve full 38-digit precision, add enable_decfloat=True to the connection URL:

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

Or using the snowflake.sqlalchemy.URL helper:

from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine
engine = create_engine(URL(
    account = 'abc123',
    user = 'testuser1',
    password = '0123456',
    database = 'testdb',
    schema = 'public',
    warehouse = 'testwh',
    enable_decfloat = True,
))

Note: DECFLOAT does not support special values (inf, -inf, NaN) unlike FLOAT.

Why is enable_decfloat not enabled by default? Enabling it sets decimal.getcontext().prec = 38, which modifies Python's thread-local decimal context and affects all Decimal operations in that thread, not just database queries. To avoid unexpected side effects on application code, the dialect emits a warning when DECFLOAT values are retrieved without full precision enabled, guiding users to opt-in explicitly.

Returning `float` Instead of `Decimal`

By default, Snowflake NUMBER / DECIMAL columns are returned as Python decimal.Decimal values. If you prefer plain float results for a column, set asdecimal=False on its numeric type:

from sqlalchemy import Column, Integer, MetaData, Table
from snowflake.sqlalchemy import NUMBER
metadata = MetaData()
prices = Table(
    "prices",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("amount", NUMBER(10, 2, asdecimal=False)),  # rows return float, not Decimal
)

The default remains asdecimal=True (Decimal), so existing behavior is unchanged.

VECTOR Data Type Support

Snowflake SQLAlchemy supports the VECTOR data type with varying element type and dimension. For more information, see the Snowflake documentation.

from sqlalchemy import Column, Integer, Float, MetaData, Table
from snowflake.sqlalchemy import VECTOR
metadata = MetaData()
t = Table('my_table', metadata,
    Column('id', Integer, primary_key=True),
    Column('int_vec', VECTOR(Integer, 20)),
    Column('float_vec', VECTOR(Float, 40)),
)
metadata.create_all(engine)

UUID Data Type Support

SQLAlchemy 2.x only. The generic UUID type does not exist in SQLAlchemy 1.4; on that version Snowflake UUID columns are reflected as NullType.

Snowflake stores UUID values internally as text strings. When a UUID column is reflected, snowflake-sqlalchemy maps it to sqlalchemy.sql.sqltypes.UUID with as_uuid=False, so query results are returned as plain hyphenated strings (e.g. "6ba7b810-9dad-11d1-80b4-00c04fd430c8") rather than uuid.UUID objects.

from sqlalchemy import Column, MetaData, Table, text
from sqlalchemy.sql.sqltypes import UUID  # SA 2.x
metadata = MetaData()
# Define a table with a UUID primary key
t = Table(
    "my_table",
    metadata,
    Column("id", UUID, primary_key=True),
)
metadata.create_all(engine)
# Reflected UUID columns come back as strings by default
with engine.connect() as conn:
    row = conn.execute(text("SELECT id FROM my_table LIMIT 1")).fetchone()
    print(type(row[0]))   # <class 'str'>
    print(row[0])         # "6ba7b810-9dad-11d1-80b4-00c04fd430c8"

If you prefer uuid.UUID objects in query results, pass as_uuid=True explicitly:

Column("id", UUID(as_uuid=True), primary_key=True)

Note that Alembic autogenerate will render the column as UUID(as_uuid=True) in generated migration files, whereas the default (as_uuid=False) renders as UUID().

Timestamp and Timezone Support

Snowflake SQLAlchemy provides three Snowflake-specific timestamp types that map directly to their Snowflake counterparts:

from sqlalchemy import Column, Integer, MetaData, Table, create_engine
from snowflake.sqlalchemy import TIMESTAMP_NTZ, TIMESTAMP_TZ, TIMESTAMP_LTZ
engine = create_engine(...)
metadata = MetaData()
t = Table('events', metadata,
    Column('id', Integer, primary_key=True),
    Column('created_at', TIMESTAMP_NTZ()),   # TIMESTAMP WITHOUT TIME ZONE
    Column('scheduled_at', TIMESTAMP_TZ()),  # TIMESTAMP WITH TIME ZONE
    Column('logged_at', TIMESTAMP_LTZ()),    # TIMESTAMP WITH LOCAL TIME ZONE
)
metadata.create_all(engine)

SQLAlchemy's generic DateTime and TIMESTAMP types also support timezone-aware columns via the timezone parameter. When timezone=True is set, the dialect emits TIMESTAMP_TZ instead of the default TIMESTAMP_NTZ:

from sqlalchemy import Column, DateTime, Integer, MetaData, Table, create_engine
from sqlalchemy.types import TIMESTAMP
engine = create_engine(...)
metadata = MetaData()
t = Table('events', metadata,
    Column('id', Integer, primary_key=True),
    Column('naive_ts', DateTime()),                 # produces TIMESTAMP_NTZ
    Column('aware_ts', DateTime(timezone=True)),    # produces TIMESTAMP_TZ
    Column('naive_ts2', TIMESTAMP()),               # produces TIMESTAMP_NTZ
    Column('aware_ts2', TIMESTAMP(timezone=True)),  # produces TIMESTAMP_TZ
)
metadata.create_all(engine)

This also applies when using pandas to_sql() with timezone-aware datetime columns, which infers DateTime(timezone=True) automatically (see #199).

Note on Time and timezones: SQLAlchemy's Time type accepts a timezone parameter, but Snowflake's TIME data type does not support time zones. Using Time(timezone=True) will compile to plain TIME and the timezone flag will have no effect. If you need to store time data with time-zone information, use a timestamp type such as TIMESTAMP_TZ or DateTime(timezone=True) instead.

Cache Column Metadata

SQLAlchemy provides the runtime inspection API to get the runtime information about the various objects. One common use case is retrieving all tables and their column metadata in a schema to construct a schema catalog. For example, alembic manages database schema migrations on top of SQLAlchemy. A typical flow (SQLAlchemy 1.4) is:

inspector = inspect(engine)
schema = inspector.default_schema_name
for table_name in inspector.get_table_names(schema):
    column_metadata = inspector.get_columns(table_name, schema)
    primary_keys = inspector.get_pk_constraint(table_name, schema)
    foreign_keys = inspector.get_foreign_keys(table_name, schema)
    ...

In this flow, running a separate query per table can be slow for large schemas. Snowflake SQLAlchemy optimises this with schema-wide cached queries and, where appropriate, fast per-table queries.

Single-Table vs Multi-Table Reflection Performance

SQLAlchemy 2.x (automatic)

SQLAlchemy 2.x distinguishes bulk reflection from single-table inspection at the framework level:

MetaData.reflect() / Table(..., autoload_with=engine) — calls get_multi_columns, get_multi_pk_constraint, get_multi_foreign_keys, and get_multi_unique_constraints. Each issues one schema-wide SHOW or information_schema query and caches the result for all tables in the schema.
inspector.get_columns(table_name) — issues a single DESC TABLE query directly against that table. This is fast and correct for all table types including temporary tables.

No configuration is needed; the routing is handled automatically by the SA 2.x dispatch layer.

Note on reflected type representations: Because inspector.get_columns() uses DESC TABLE, reflected types always include Snowflake's resolved default sizes (e.g. BINARY(8388608) instead of BINARY, VARCHAR(16777216) instead of VARCHAR). The type objects are functionally identical; only str() output differs. Use isinstance() checks rather than string comparison for type introspection.

from sqlalchemy import MetaData, inspect, create_engine
engine = create_engine('snowflake://...')
# SA 2.x: one schema-wide query per metadata type, all tables cached at once
metadata = MetaData()
metadata.reflect(bind=engine, schema='public')
# SA 2.x: direct DESC TABLE, no schema-wide query issued
inspector = inspect(engine)
columns = inspector.get_columns('my_table', schema='public')

Per-table optimisation

get_pk_constraint, get_unique_constraints, get_foreign_keys, get_indexes, and get_columns automatically use per-table queries (SHOW … IN TABLE, DESC TABLE) for single-table Inspector calls (e.g. Inspector.get_pk_constraint(), pandas.read_sql_table()). MetaData.reflect() continues to use the schema-wide get_multi_* hooks, which issue one query per reflection pass.

Performance Implications

For schemas with many tables (100+), schema-wide queries issued once during MetaData.reflect() are far more efficient than per-table queries in a loop:

Schema-wide SHOW PRIMARY KEYS IN SCHEMA (all tables): < 1 second
Per-table loop over 1 000 tables: 1 000+ round-trips

For single-table inspection via Inspector, per-table queries (DESC TABLE, SHOW … IN TABLE) are faster than fetching the entire schema.

Best Practices

For bulk reflection: use metadata.reflect() — schema-wide queries are issued once and cached.
For single-table inspection: use inspector.get_columns() / inspector.get_pk_constraint() etc. — per-table queries are used automatically.
For very large schemas: reflect only the tables you need:

metadata.reflect(bind=engine, schema='public', only=['table1', 'table2'])

Cross-Database Reflection

Snowflake SQLAlchemy supports reflecting tables from different databases using the database.schema notation in the schema parameter. This allows you to work with tables from multiple databases in a single session without using raw SQL.

To reflect a table from a different database, use the database.schema notation:

from sqlalchemy import create_engine, MetaData, Table, select
# Connect to database_a
engine = create_engine('snowflake://user:pass@account/database_a')
metadata = MetaData()
# Reflect a table from database_b using database.schema notation
bananas = Table(
    'bananas',
    metadata,
    schema='database_b.schema_b',  # Cross-database schema
    autoload_with=engine
)
# Reflect a table from the current database
apples = Table(
    'apples',
    metadata,
    schema='schema_a',  # Uses current database (database_a)
    autoload_with=engine
)
# Create a cross-database join
stmt = select(apples, bananas).join(
    bananas,
    apples.c.id == bananas.c.id
)
# The generated SQL will use fully-qualified names for cross-database tables:
# SELECT ... FROM schema_a.apples
# JOIN database_b.schema_b.bananas ON ...

You can also reflect all tables from a different database:

metadata.reflect(
    bind=engine,
    schema='database_b.schema_b'
)

Note about schema names containing dots: If your Snowflake schema name literally contains a dot character (e.g., created with CREATE SCHEMA "my.schema"), you must quote it in the Python string:

# For schema literally named "my.schema" in database_b
table = Table('mytable', metadata, schema='database_b."my.schema"', autoload_with=engine)

The SQL compilation layer already treats unquoted dots as separators. This cross-database reflection feature makes the reflection layer consistent with that existing behavior.

Reflecting Large Schemas (10,000+ objects)

Snowflake SHOW commands return at most 10,000 rows per call. Reflection helpers that list objects in a schema issue SHOW ... IN SCHEMA, so before this was addressed a schema (or database) with more than 10,000 objects would fail with an error like:

090201 (22000): SHOW ... returned more than the maximum of 10000 results.
Use LIMIT to reduce the number of results.

Snowflake SQLAlchemy now transparently pages through these object-listing SHOW commands using LIMIT <n> FROM '<last_name>', so reflection works regardless of object count. No code change is required on your side:

from sqlalchemy import create_engine, inspect
engine = create_engine("snowflake://<user>:<password>@<account>/<db>/<schema>")
inspector = inspect(engine)
# All of these page automatically and no longer fail past 10,000 objects:
inspector.get_table_names(schema="my_schema")     # SHOW TABLES IN SCHEMA
inspector.get_view_names(schema="my_schema")      # SHOW VIEWS IN
inspector.get_sequence_names(schema="my_schema")  # SHOW SEQUENCES IN SCHEMA
inspector.get_schema_names()                       # SHOW SCHEMAS
inspector.get_temp_table_names(schema="my_schema")  # SHOW TABLES IN SCHEMA (temp)

MetaData.reflect() / Table(..., autoload_with=engine) benefit as well, since they call these helpers internally.

How paging works

Each command is fetched in pages ordered lexicographically by object name; each page after the first seeks past the previous page's last name:

SHOW TABLES IN SCHEMA "MY_DB"."MY_SCHEMA" LIMIT 10000
SHOW TABLES IN SCHEMA "MY_DB"."MY_SCHEMA" LIMIT 10000 FROM '<last name of previous page>'
-- ...repeated until a page returns fewer than 10000 rows

The page size defaults to Snowflake's 10,000 maximum. It is exposed as SnowflakeDialect._SHOW_TABLES_PAGE_SIZE (clamped to 1–10,000) primarily so tests can exercise the paging loop with a small value; you normally do not need to change it.

Known limitations

Only the object-listing SHOW ... IN [SCHEMA] commands are paged (tables, views, temp tables, schemas, sequences). The schema-wide constraint/index commands are not yet paged and can still hit the 10,000-row cap on very large schemas when using the bulk MetaData.reflect() path:

SHOW PRIMARY KEYS IN SCHEMA
SHOW UNIQUE KEYS IN SCHEMA
SHOW IMPORTED KEYS IN SCHEMA
SHOW INDEXES IN SCHEMA

These lack a name column to page by and generally do not support LIMIT ... FROM, so they require a different approach (per-table fallback) tracked separately. Reflecting a single table's primary key, unique/foreign keys, or indexes is unaffected — those use the bounded SHOW ... IN TABLE form. Single-object lookups such as get_view_definition() (SHOW VIEWS LIKE ...) are likewise unaffected.

VARIANT, ARRAY and OBJECT Support

Snowflake SQLAlchemy supports fetching VARIANT, ARRAY and OBJECT data types. All types are converted into str in Python so that you can convert them to native data types using json.loads.

This example shows how to create a table including VARIANT, ARRAY, and OBJECT data type columns.

from snowflake.sqlalchemy import (VARIANT, ARRAY, OBJECT)
t = Table('my_semi_strucutred_datatype_table', metadata,
    Column('va', VARIANT),
    Column('ob', OBJECT),
    Column('ar', ARRAY))
metdata.create_all(engine)

In order to retrieve VARIANT, ARRAY, and OBJECT data type columns and convert them to the native Python data types, fetch data and call the json.loads method as follows:

import json
connection = engine.connect()
results = connection.execute(select([t])
row = results.fetchone()
data_variant = json.loads(row[0])
data_object  = json.loads(row[1])
data_array   = json.loads(row[2])

Writing Python dicts/lists to VARIANT, OBJECT and ARRAY

Snowflake does not accept a bound Python dict/list for a semi-structured column, and it rejects a function expression such as PARSE_JSON(?) inside an INSERT ... VALUES clause. Binding a raw dict raises 255001: Binding data in type (dict) is not supported, and binding a JSON string fails with an expecting VARIANT but got VARCHAR type mismatch.

The supported pattern is INSERT ... SELECT (or UPDATE) with PARSE_JSON applied to a JSON string:

import json
from sqlalchemy import select, literal, func
# INSERT ... SELECT PARSE_JSON(?)
connection.execute(
    t.insert().from_select(
        ["id", "va"],
        select(literal(1), func.parse_json(json.dumps({"a": 1, "b": [2, 3]}))),
    )
)
# UPDATE ... SET va = PARSE_JSON(?)
connection.execute(
    t.update().where(t.c.id == 1).values(va=func.parse_json(json.dumps({"a": 2})))
)

Use func.parse_json(json.dumps(value)) anywhere you would otherwise bind a dict/list to a VARIANT, OBJECT, or ARRAY column.

Automatic JSON handling with `enable_structured_type_json`

The manual PARSE_JSON / json.loads steps above can be handled automatically by opting in to enable_structured_type_json — either as a create_engine argument or a URL parameter:

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL
engine = create_engine(URL(account="myaccount", user="me", password="secret"),
                       enable_structured_type_json=True)
# or: create_engine("snowflake://...?enable_structured_type_json=true")

When enabled, for semi-structured (untyped) VARIANT, OBJECT and ARRAY columns:

Writing a native dict/list serializes it and wraps it in PARSE_JSON. Because Snowflake rejects functions in a VALUES clause, inserts are rendered as INSERT ... SELECT (multi-row inserts become SELECT ... UNION ALL SELECT ...); UPDATE renders SET col = PARSE_JSON(...).
Reading deserializes the JSON text Snowflake returns back into native Python (dict/list).

with engine.begin() as conn:
    conn.execute(t.insert().values(id=1, va={"a": 1, "b": [2, 3]}))          # single row
    conn.execute(t.insert().values([{"id": 2, "va": {"a": 2}},
                                    {"id": 3, "va": {"a": 3}}]))             # multi-row (UNION ALL)
with engine.connect() as conn:
    row = conn.execute(select(t.c.va).where(t.c.id == 1)).one()
    assert row.va == {"a": 1, "b": [2, 3]}   # dict, not a JSON string

Notes:

The flag is off by default, so existing code that reads raw JSON strings or writes pre-serialized values with the PARSE_JSON pattern above is unaffected.
Typed/structured columns (OBJECT(...) with fields, ARRAY(<type>), MAP(...)) keep their native connector handling and are not wrapped in PARSE_JSON.
The engine-level json_serializer / json_deserializer (below) are used when provided.
RETURNING is not supported for inserts into semi-structured columns while this flag is on.
executemany (2+ parameter sets) is not supported while this flag is on. Passing a list of two or more parameter dicts — conn.execute(t.insert(), [row1, row2, ...]), and equivalently ORM bulk flushes that batch multiple new objects in one commit — raises 252001: Failed to rewrite multi-row insert. The driver tries to fold the rows into a single VALUES (...) clause, which the INSERT ... SELECT PARSE_JSON(...) rewrite does not have. Use a single-row insert, a single parameter dict (conn.execute(t.insert(), {...})), or the multi-row insert().values([...]) form shown above (rendered as SELECT ... UNION ALL).
For large data volumes, prefer staging + COPY INTO (or write_pandas) over the PARSE_JSON write path. Because PARSE_JSON must be inlined into the statement text, both per-row and UNION ALL rewrites are bounded by Snowflake's ~1 MB statement-size limit and are inefficient for bulk loads.
literal_binds is not supported for semi-structured writes. Compiling an insert/update of a dict/list with compile_kwargs={"literal_binds": True} raises, because inlining the JSON as a SQL literal is unsafe (Snowflake unescapes backslashes in single-quoted literals before PARSE_JSON runs, so escaping cannot both preserve the data and prevent literal breakout). Execute with bound parameters (the default) instead; a None value still renders as SQL NULL.

Reading keys and elements with subscript access

VARIANT, OBJECT, ARRAY and MAP columns support Python subscript syntax in queries. col["key"] and col[index] compile to Snowflake's native bracket accessor, and access chains can be nested:

from sqlalchemy import select
# SELECT tbl.ob['name'], tbl.ar[0], tbl.mp['k'], tbl.ob['address']['city'] FROM tbl
select(
    t.c.ob["name"],
    t.c.ar[0],
    t.c.mp["k"],
    t.c.ob["address"]["city"],
)

Subscript expressions can also be used in WHERE, ORDER BY, and other clauses:

# SELECT tbl.id FROM tbl WHERE tbl.ob['status'] = 'active'
select(t.c.id).where(t.c.ob["status"] == "active")

The extracted value is itself semi-structured; cast it (for example t.c.ob["age"].cast(Integer)) when you need a specific SQL type.

A few limitations follow from Snowflake's bracket accessor:

Negative indices are not supported. col[-1] compiles as-is, but Snowflake raises a runtime error for negative array indices — use a non-negative index.
A key/index bound as a parameter needs client-side substitution. With the default pyformat paramstyle the driver substitutes the bound value before the statement reaches Snowflake, so col[some_bindparam] works. Under a positional qmark paramstyle the bracket accessor cannot take a placeholder; pass the key/index as a literal (for example col["name"] or col[0]) instead of a bound parameter.

Customizing JSON serialization (`json_serializer` / `json_deserializer`)

Snowflake SQLAlchemy accepts the standard SQLAlchemy engine-level json_serializer and json_deserializer callables, matching the built-in dialects. Use them to plug in a custom JSON encoder/decoder — for example to preserve Decimal precision instead of parsing floats:

import json
import decimal
import simplejson
from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL
engine = create_engine(
    URL(account="myaccount", user="me", password="secret"),
    json_serializer=lambda obj: simplejson.dumps(obj, use_decimal=True),
    json_deserializer=lambda value: json.loads(value, parse_float=decimal.Decimal),
)

Structured Data Types Support

This module defines custom SQLAlchemy types for Snowflake structured data, specifically for Iceberg tables. The types —MAP, OBJECT, and ARRAY— allow you to store complex data structures in your SQLAlchemy models. For detailed information, refer to the Snowflake Structured data types documentation.

MAP

The MAP type represents a collection of key-value pairs, where each key and value can have different types.

Key Type: The type of the keys (e.g., TEXT, NUMBER).
Value Type: The type of the values (e.g., TEXT, NUMBER).
Not Null: Whether NULL values are allowed (default is False).

Example Usage

IcebergTable(
    table_name,
    metadata,
    Column("id", Integer, primary_key=True),
    Column("map_col", MAP(NUMBER(10, 0), TEXT(16777216))),
    external_volume="external_volume",
    base_location="base_location",
)

OBJECT

The OBJECT type represents a semi-structured object with named fields. Each field can have a specific type, and you can also specify whether each field is nullable.

Items Types: A dictionary of field names and their types. The type can optionally include a nullable flag (True for not nullable, False for nullable, default is False).

Example Usage

IcebergTable(
    table_name,
    metadata,
    Column("id", Integer, primary_key=True),
    Column(
        "object_col",
        OBJECT(key1=(TEXT(16777216), False), key2=(NUMBER(10, 0), False)),
        OBJECT(key1=TEXT(16777216), key2=NUMBER(10, 0)), # Without nullable flag
    ),
    external_volume="external_volume",
    base_location="base_location",
)

ARRAY

The ARRAY type represents an ordered list of values, where each element has the same type. The type of the elements is defined when creating the array.

Value Type: The type of the elements in the array (e.g., TEXT, NUMBER).
Not Null: Whether NULL values are allowed (default is False).

Example Usage

IcebergTable(
    table_name,
    metadata,
    Column("id", Integer, primary_key=True),
    Column("array_col", ARRAY(TEXT(16777216))),
    external_volume="external_volume",
    base_location="base_location",
)

CLUSTER BY Support

Snowflake SQLAchemy supports the CLUSTER BY parameter for tables. For information about the parameter, see :doc:/sql-reference/sql/create-table.

This example shows how to create a table with two columns, id and name, as the clustering keys:

t = Table('myuser', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    snowflake_clusterby=['id', 'name', text('id > 5')], ...
)
metadata.create_all(engine)

ORM / Declarative usage

For declarative models, use the SnowflakeTable construct, which accepts a cluster_by argument (plain column names and/or text() expressions) — see #313:

from sqlalchemy import Column, Integer, String, text
from snowflake.sqlalchemy.sql.custom_schema import SnowflakeTable
my_table = SnowflakeTable(
    "my_table",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    cluster_by=["id", text("date_trunc('hour', ts)")],
)

CLUSTER BY is a storage optimization applied at DDL time and does not change query semantics — you query a clustered table exactly as any other table. To filter on a clustering expression, build it in the WHERE clause as usual:

from datetime import datetime, timedelta
from sqlalchemy import func, select
stmt = select(my_table).where(
    func.date_trunc("hour", my_table.c.ts) > datetime.utcnow() - timedelta(hours=1)
)

Row Access Policy Support

Snowflake SQLAlchemy can attach an existing row access policy to a table at CREATE TABLE time. Use the SnowflakeTable construct and pass row_access_policy as a (policy_name, [columns]) tuple:

from sqlalchemy import Column, Integer, MetaData, String
from snowflake.sqlalchemy.sql.custom_schema import SnowflakeTable
metadata = MetaData()
my_table = SnowflakeTable(
    "my_table",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("geom", String),
    Column("name", String),
    row_access_policy=("my_policy", ["geom", "name"]),
)
metadata.create_all(engine)
# CREATE TABLE my_table (...) WITH ROW ACCESS POLICY my_policy ON (geom, name)

You can also pass an explicit RowAccessPolicyOption instead of a tuple:

from snowflake.sqlalchemy import RowAccessPolicyOption
row_access_policy = RowAccessPolicyOption("my_policy", ["geom", "name"])

The policy name may be fully qualified. A string is parsed as [[database.]schema.]name (each part is quoted independently, and only when it requires quoting or was already quoted), or you can pass a FQN explicitly:

from snowflake.sqlalchemy import FQN
# both render: WITH ROW ACCESS POLICY gov_db.security.my_policy ON (geom, name)
row_access_policy = ("gov_db.security.my_policy", ["geom", "name"])
row_access_policy = RowAccessPolicyOption(
    FQN.from_string("gov_db.security.my_policy"), ["geom", "name"]
)
# a quoted part is preserved verbatim; unquoted parts stay bare:
# renders: WITH ROW ACCESS POLICY gov_db."Sec".my_policy ON (geom, name)
row_access_policy = ('gov_db."Sec".my_policy', ["geom", "name"])

The policy itself must already exist in Snowflake. The option is opt-in and additive — tables declared without row_access_policy are unaffected.

RELY Constraint Property

Snowflake does not enforce PRIMARY KEY, UNIQUE, or FOREIGN KEY constraints — they are metadata only. The RELY property tells the query optimizer it may trust such a constraint for rewrites like join elimination, even though it isn't enforced. Set the opt-in snowflake_rely=True dialect keyword on a constraint to emit it:

from sqlalchemy import (
    Column, ForeignKeyConstraint, Integer, MetaData,
    PrimaryKeyConstraint, String, Table, UniqueConstraint,
)
metadata = MetaData()
dim = Table(
    "dim", metadata,
    Column("id", Integer),
    Column("email", String),
    PrimaryKeyConstraint("id", snowflake_rely=True),
    UniqueConstraint("email", snowflake_rely=True),
)
# ... PRIMARY KEY (id) RELY ... UNIQUE (email) RELY
fact = Table(
    "fact", metadata,
    Column("dim_id", Integer),
    ForeignKeyConstraint(["dim_id"], ["dim.id"], snowflake_rely=True),
)
# ... FOREIGN KEY(dim_id) REFERENCES dim (id) RELY

Warning: RELY is an unverified promise. Snowflake trusts it without checking the data. If the data actually violates the constraint (duplicate keys, orphan foreign keys), the optimizer may rewrite queries based on a false assumption and return incorrect results. Only set snowflake_rely=True when you are certain the data satisfies the constraint. It defaults to off, so existing DDL is unaffected.

Alembic Support

Alembic is a database migration tool on top of SQLAlchemy. Snowflake SQLAlchemy works by adding the following code to alembic/env.py so that Alembic can recognize Snowflake SQLAlchemy.

from alembic.ddl.impl import DefaultImpl
class SnowflakeImpl(DefaultImpl):
    __dialect__ = 'snowflake'

See Alembic Documentation for general usage.

Key Pair Authentication Support

Snowflake SQLAlchemy supports key pair authentication by leveraging its Snowflake Connector for Python underpinnings. See Using Key Pair Authentication for steps to create the private and public keys.

The private key parameter is passed through connect_args as follows:

...
from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives.asymmetric import rsa
from cryptography.hazmat.primitives.asymmetric import dsa
from cryptography.hazmat.primitives import serialization
with open("rsa_key.p8", "rb") as key:
    p_key= serialization.load_pem_private_key(
        key.read(),
        password=os.environ['PRIVATE_KEY_PASSPHRASE'].encode(),
        backend=default_backend()
    )
pkb = p_key.private_bytes(
    encoding=serialization.Encoding.DER,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption())
engine = create_engine(URL(
    account='abc123',
    user='testuser1',
    ),
    connect_args={
        'private_key': pkb,
        },
    )

Where PRIVATE_KEY_PASSPHRASE is a passphrase to decrypt the private key file, rsa_key.p8.

Currently a private key parameter is not accepted by the snowflake.sqlalchemy.URL method.

PrivateLink and Account Name Handling

When connecting over AWS PrivateLink, the host uses the .privatelink notation. Snowflake SQLAlchemy normalizes the account name from the host using the Snowflake Connector's own account parsing, so behavior matches the driver for all supported notations:

Host notation	Derived account
`myaccount`	`myaccount`
`myaccount.us-east-1`	`myaccount`
`orgname-account_name.us-east-1` (regional)	`orgname-account_name`
`myaccount.us-east-1.privatelink`	`myaccount`
`orgname-account_name.privatelink` (regionless)	`orgname-account_name`
`myaccount-<external_id>.global`	`myaccount`

Account names that contain a dash (org-style orgname-account_name, for example over regionless PrivateLink) are preserved intact, so the account name in a Key Pair / JWT token matches the account used to connect. No extra configuration is required.

SSO Authentication (Okta, Azure AD, etc.)

Snowflake SQLAlchemy supports federated/SSO authentication by forwarding the authenticator parameter to the Snowflake Connector for Python. No dialect-specific configuration is required.

For native Okta, pass your Okta endpoint URL as the authenticator:

from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL
engine = create_engine(
    URL(
        account="abc123",
        user="testuser1",
        database="testdb",
        schema="public",
    ),
    connect_args={"authenticator": "https://your-okta-domain.okta.com"},
)

For browser-based SSO (any configured identity provider, including Azure AD), use externalbrowser, which opens your default browser to complete the login:

engine = create_engine(
    URL(account="abc123", user="testuser1"),
    connect_args={"authenticator": "externalbrowser"},
)

See Snowflake's federated authentication and SSO documentation for identity-provider setup instructions.

Boolean Column Comparisons (`IS TRUE` / `IS FALSE`)

SQLAlchemy's idiomatic col.is_(True) / col.is_(False) compile to IS TRUE / IS FALSE, which Snowflake does not support (Snowflake only allows IS [NOT] NULL). Using them raises a syntax error (see #680):

from sqlalchemy import select
# ❌ Fails — Snowflake has no IS TRUE / IS FALSE:
select(t).where(t.c.flag.is_(True))
# snowflake.connector.errors.ProgrammingError: 001003 (42000):
#   SQL compilation error: syntax error ... unexpected 'IS'.

Use equality against SQLAlchemy's true() / false() helpers instead:

from sqlalchemy import true, false, select
select(t).where(t.c.flag == true())     # WHERE flag = TRUE
select(t).where(t.c.flag == false())    # WHERE flag = FALSE

Note the NULL semantics differ from IS TRUE/IS FALSE: flag = TRUE yields NULL for a NULL row (which filters it out), whereas flag IS TRUE would yield FALSE. If you need the strict IS TRUE behavior, add an explicit null check:

from sqlalchemy import and_, true
select(t).where(and_(t.c.flag.isnot(None), t.c.flag == true()))
# WHERE flag IS NOT NULL AND flag = TRUE

Merge Command Support

Snowflake SQLAlchemy supports upserting with its MergeInto custom expression. See Merge for full documentation.

Use it as follows:

from sqlalchemy.orm import sessionmaker
from sqlalchemy import MetaData, create_engine
from snowflake.sqlalchemy import MergeInto
engine = create_engine(db.url, echo=False)
session = sessionmaker(bind=engine)()
connection = engine.connect()
meta = MetaData()
meta.reflect(bind=session.bind)
t1 = meta.tables['t1']
t2 = meta.tables['t2']
merge = MergeInto(target=t1, source=t2, on=t1.c.t1key == t2.c.t2key)
merge.when_matched_then_delete().where(t2.c.marked == 1)
merge.when_matched_then_update().where(t2.c.isnewstatus == 1).values(val = t2.c.newval, status=t2.c.newstatus)
merge.when_matched_then_update().values(val=t2.c.newval)
merge.when_not_matched_then_insert().values(val=t2.c.newval, status=t2.c.newstatus)
connection.execute(merge)

Using a `VALUES` source (no staging table required)

MergeInto's source accepts any SQLAlchemy Selectable, so you can merge from an inline VALUES list (wrapped in a subquery) instead of an existing table:

from sqlalchemy import values, column, Integer, String, select
src = select(
    values(
        column("id", Integer),
        column("status", String),
        name="src",
    ).data([(1, "active"), (2, "inactive")])
).subquery()
merge = MergeInto(target=t1, source=src, on=t1.c.id == src.c.id)
merge.when_matched_then_update().values(status=src.c.status)
merge.when_not_matched_then_insert().values(id=src.c.id, status=src.c.status)
connection.execute(merge)

Multi-Table Insert (`INSERT ALL` / `INSERT FIRST`)

Snowflake SQLAlchemy supports Snowflake's multi-table INSERT ALL / INSERT FIRST statements through the InsertMulti custom command. Build it from a source SELECT and add one or more targets.

Unconditional insert with .into(...):

from sqlalchemy import select
from snowflake.sqlalchemy import InsertMulti
stmt = (
    InsertMulti(select(source_table))
    .into(target1, columns=["id", "name", "value"])
    .into(target2, columns=["id", "name", "value"])
)
connection.execute(stmt)
# INSERT ALL INTO target1 (...) INTO target2 (...) SELECT ... FROM source_table

Each target may map specific source columns/expressions via values; when omitted, the subquery output columns are used positionally.

Conditional insert with .when(...) / .else_(...) (use first=True for INSERT FIRST, overwrite=True for INSERT OVERWRITE):

stmt = select(src.c.id, src.c.name, src.c.active)
insert = (
    InsertMulti(stmt)
    .when(stmt.selected_columns.active, active_users,
          values=[stmt.selected_columns.id, stmt.selected_columns.name])
    .when(~stmt.selected_columns.active, inactive_users,
          values=[stmt.selected_columns.id, stmt.selected_columns.name])
    .else_(other_users)
)
connection.execute(insert)

InsertMulti is a new, additive construct; existing code is unaffected.

Bulk Insert Optimization for ORM Models

When using Session.bulk_save_objects() with models that have nullable optional columns, SQLAlchemy groups objects into separate INSERT batches based on each object's set of non-None column keys. If some objects were constructed without supplying every nullable column, they produce different key sets and SQLAlchemy emits O(N) INSERT statements instead of a single executemany batch.

Snowflake SQLAlchemy provides two components that together solve this problem:

SnowflakeBase (SQLAlchemy 2.x only) — a DeclarativeBase subclass whose constructor pre-populates every plain-nullable column with None (or its scalar Python default) at construction time, so all instances share the same column-key set regardless of which kwargs the caller supplied.
snowflake_declarative_base() — a factory function compatible with both SQLAlchemy 1.4 and 2.x that produces a declarative base with the same pre-population behaviour.
SnowflakeSession — a Session subclass that passes render_nulls=True to the internal bulk-save call, preventing pre-populated None values from being stripped before grouping. Must be used together with SnowflakeBase or snowflake_declarative_base() for full effect.

SQLAlchemy 2.x example:

from sqlalchemy import Column, Integer, String, create_engine
from snowflake.sqlalchemy import SnowflakeBase, SnowflakeSession
class MyModel(SnowflakeBase):
    __tablename__ = "my_model"
    id = Column(Integer, primary_key=True)
    name = Column(String)       # nullable, no default
    status = Column(String, default="active")  # scalar default
engine = create_engine("snowflake://...")
SnowflakeBase.metadata.create_all(engine)
session = SnowflakeSession(bind=engine)
# All objects share the same column-key set — emits a single executemany INSERT
session.bulk_save_objects([
    MyModel(id=1),
    MyModel(id=2, name="foo"),
    MyModel(id=3, status="inactive"),
])
session.commit()

SQLAlchemy 1.4 and 2.x compatible example:

from sqlalchemy import Column, Integer, String, create_engine
from snowflake.sqlalchemy import snowflake_declarative_base, SnowflakeSession
Base = snowflake_declarative_base()
class MyModel(Base):
    __tablename__ = "my_model"
    id = Column(Integer, primary_key=True)
    name = Column(String)
    status = Column(String, default="active")
engine = create_engine("snowflake://...")
Base.metadata.create_all(engine)
session = SnowflakeSession(bind=engine)
session.bulk_save_objects([
    MyModel(id=1),
    MyModel(id=2, name="foo"),
])
session.commit()

Notes:

SnowflakeBase is only available in SQLAlchemy 2.x. Use snowflake_declarative_base() when your code must run on both SA 1.4 and 2.x.
Columns with server_default, callable Python defaults (default=fn), or SQL-expression defaults (default=func.now()) are intentionally left absent from pre-population. Objects that differ on such columns may still be placed in separate INSERT batches — this is the same behaviour as stock SQLAlchemy.
SnowflakeSession alone (without the matching base class) is not sufficient: the base class is required to unify the column-key sets before SnowflakeSession can batch them together.

Bulk Array Binding for Large `executemany` Inserts

For large batch inserts (e.g. Connection.execute(insert(t), [row1, row2, ...]), session.bulk_insert_mappings(), or any cursor.executemany-backed path), the Snowflake Connector for Python can upload the bind values to a temporary internal stage as CSV and run a single insert that reads them from the stage, instead of embedding a large inline VALUES payload. This "bulk array binding" is faster and far more scalable for high row counts.

This optimization is only used when the paramstyle is qmark (or numeric). The dialect defaults to pyformat (to match the connector default), so you must switch both the SQLAlchemy engine and the underlying connector connection to qmark — they must agree, or binding will fail:

from sqlalchemy import create_engine
engine = create_engine(
    "snowflake://<account>/<db>/<schema>",
    paramstyle="qmark",                 # SQLAlchemy renders positional "?" placeholders
    connect_args={
        "paramstyle": "qmark",          # connector uses qmark -> enables stage array binding
    },
)

Optionally tune when the stage upload kicks in with the connector session parameter CLIENT_STAGE_ARRAY_BINDING_THRESHOLD (total bind size in bytes; 0 disables the optimization):

engine = create_engine(
    "snowflake://<account>/<db>/<schema>",
    paramstyle="qmark",
    connect_args={
        "paramstyle": "qmark",
        "CLIENT_STAGE_ARRAY_BINDING_THRESHOLD": 65536,
    },
)

Notes:

Both paramstyle settings must match. Setting only one side causes a placeholder/binding mismatch — the engine paramstyle controls how SQLAlchemy renders the SQL, while the connect_args["paramstyle"] controls how the connector processes the parameters.
qmark applies engine-wide: every statement compiled by that engine uses positional binds.
Below the threshold (or for smaller batches) the connector sends the binds inline as usual; the staged path is transparent to your code either way.

CopyIntoStorage Support

Snowflake SQLAlchemy supports saving tables/query results into different stages, as well as into Azure Containers and AWS buckets with its custom CopyIntoStorage expression. See Copy into for full documentation.

Use it as follows:

from sqlalchemy.orm import sessionmaker
from sqlalchemy import MetaData, create_engine
from snowflake.sqlalchemy import CopyIntoStorage, AWSBucket, CSVFormatter
engine = create_engine(db.url, echo=False)
session = sessionmaker(bind=engine)()
connection = engine.connect()
meta = MetaData()
meta.reflect(bind=session.bind)
users = meta.tables['users']
copy_into = CopyIntoStorage(from_=users,
                            into=AWSBucket.from_uri('s3://my_private_backup').encryption_aws_sse_kms('1234abcd-12ab-34cd-56ef-1234567890ab'),
                            formatter=CSVFormatter().null_if(['null', 'Null']))
connection.execute(copy_into)

Securing cloud credentials in logs

When a CopyIntoStorage / CreateStage uses inline credentials (AWSBucket.credentials(...), AzureContainer.credentials(...), or any encryption_*_cse(master_key)), those secrets are emitted as literal values in the compiled SQL — that is how Snowflake receives them. If the SQLAlchemy engine log is enabled (create_engine(..., echo=True) or the sqlalchemy.engine logger at INFO/DEBUG), the full statement, secrets included, is written to your log sink. That logger is independent of the Snowflake connector's secret masking.

Preferred: use a named STORAGE_INTEGRATION instead of inline credentials, so no secret ever appears in the SQL:

copy_into = CopyIntoStorage(from_=users, into=AWSBucket.from_uri('s3://my_private_backup'))
copy_into.storage_integration('my_s3_integration')

When inline credentials are unavoidable, attach the redaction filter so the secret values are masked (***) in log output. Attach it to the handler that actually emits the records (handler filters run on propagated records; logger filters do not):
```
import logging
from snowflake.sqlalchemy import add_secret_redaction_filter
handler = logging.StreamHandler()
add_secret_redaction_filter(handler)
logging.getLogger("sqlalchemy.engine").addHandler(handler)
```
redact_secrets(text) is also exported for redacting a statement string directly. Note that object repr() (AWSBucket, AzureContainer, CopyIntoStorage, ...) already masks these secrets.

Creating a named file format

Use CreateFileFormat together with a formatter to emit a CREATE FILE FORMAT statement. It supports OR REPLACE (replace_if_exists=True), IF NOT EXISTS (if_not_exists=True, which is mutually exclusive with OR REPLACE) and an optional comment:

from snowflake.sqlalchemy import CreateFileFormat, CSVFormatter
create_format = CreateFileFormat(
    format_name="my_db.my_schema.my_csv_format",
    formatter=CSVFormatter().field_delimiter(","),
    if_not_exists=True,
    comment="CSV format for ingestion",
)
connection.execute(create_format)
# CREATE FILE FORMAT IF NOT EXISTS my_db.my_schema.my_csv_format
#     TYPE='csv' FIELD_DELIMITER = ',' COMMENT = 'CSV format for ingestion'

Iceberg Table with Snowflake Catalog support

Snowflake SQLAlchemy supports Iceberg Tables with the Snowflake Catalog, along with various related parameters. For detailed information about Iceberg Tables, refer to the Snowflake CREATE ICEBERG documentation.

To create an Iceberg Table using Snowflake SQLAlchemy, you can define the table using the SQLAlchemy Core syntax as follows:

table = IcebergTable(
    "myuser",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    external_volume=external_volume_name,
    base_location="my_iceberg_table",
    as_query="SELECT * FROM table"
)

Alternatively, you can define the table using a declarative approach:

class MyUser(Base):
    __tablename__ = "myuser"
    @classmethod
    def __table_cls__(cls, name, metadata, *arg, **kw):
        return IcebergTable(name, metadata, *arg, **kw)
    __table_args__ = {
        "external_volume": "my_external_volume",
        "base_location": "my_iceberg_table",
        "as_query": "SELECT * FROM table",
    }
    id = Column(Integer, primary_key=True)
    name = Column(String)

Hybrid Table support

Snowflake SQLAlchemy supports Hybrid Tables with indexes. For detailed information, refer to the Snowflake CREATE HYBRID TABLE documentation.

To create a Hybrid Table and add an index, you can use the SQLAlchemy Core syntax as follows:

table = HybridTable(
    "myuser",
    metadata,
    Column("id", Integer, primary_key=True),
    Column("name", String),
    Index("idx_name", "name")
)

Alternatively, you can define the table using the declarative approach:

class MyUser(Base):
    __tablename__ = "myuser"
    @classmethod
    def __table_cls__(cls, name, metadata, *arg, **kw):
        return HybridTable(name, metadata, *arg, **kw)
    __table_args__ = (
        Index("idx_name", "name"),
    )
    id = Column(Integer, primary_key=True)
    name = Column(String)

Dynamic Tables support

Snowflake SQLAlchemy supports Dynamic Tables. For detailed information, refer to the Snowflake CREATE DYNAMIC TABLE documentation.

To create a Dynamic Table, you can use the SQLAlchemy Core syntax as follows:

dynamic_test_table_1 = DynamicTable(
    "dynamic_MyUser",
    metadata,
    Column("id", Integer),
    Column("name", String),
    target_lag=(1, TimeUnit.HOURS),  # Additionally, you can use SnowflakeKeyword.DOWNSTREAM
    warehouse='test_wh',
    refresh_mode=SnowflakeKeyword.FULL,
    as_query="SELECT id, name from MyUser;"
)

Alternatively, you can define a table without columns using the SQLAlchemy select() construct:

dynamic_test_table_1 = DynamicTable(
    "dynamic_MyUser",
    metadata,
    target_lag=(1, TimeUnit.HOURS),
    warehouse='test_wh',
    refresh_mode=SnowflakeKeyword.FULL,
    as_query=select(MyUser.id, MyUser.name)
)

Notes

Defining a primary key in a Dynamic Table is not supported, meaning declarative tables don’t support Dynamic Tables.
When using the as_query parameter with a string, you must explicitly define the columns. However, if you use the SQLAlchemy select() construct, you don’t need to explicitly define the columns.
Direct data insertion into Dynamic Tables is not supported.

Verifying Package Signatures

To ensure the authenticity and integrity of the Python package, follow the steps below to verify the package signature using cosign.

Steps to verify the signature:

Install cosign:
- This example is using golang installation: installing-cosign-with-go
Download the file from the repository like pypi:
- https://pypi.org/project/snowflake-sqlalchemy/#files
Download the signature files from the release tag, replace the version number with the version you are verifying:
- https://github.com/snowflakedb/snowflake-sqlalchemy/releases/tag/v1.7.3

Verify signature:

# replace the version number with the version you are verifying
./cosign verify-blob snowflake_sqlalchemy-1.7.3-py3-none-any.whl  \
--certificate snowflake_sqlalchemy-1.7.3-py3-none-any.whl.crt \
--certificate-identity https://github.com/snowflakedb/snowflake-sqlalchemy/.github/workflows/python-publish.yml@refs/tags/v1.7.3 \
--certificate-oidc-issuer https://token.actions.githubusercontent.com \
--signature snowflake_sqlalchemy-1.7.3-py3-none-any.whl.sig
Verified OK

Support

Feel free to file an issue or submit a PR here for general cases. For contributor guidelines, development setup, and how to run the tests, see CONTRIBUTING.md. For official support, contact Snowflake support at: https://community.snowflake.com/s/article/How-To-Submit-a-Support-Case-in-Snowflake-Lodge

Known Limitations

Identity columns as primary keys

Using SQLAlchemy's Identity() construct on a primary key column is not compatible with the SQLAlchemy ORM when targeting Snowflake.

Why it fails: After an INSERT, the ORM must retrieve the generated primary key to populate the in-memory object. SQLAlchemy supports two mechanisms for this — RETURNING (not available in Snowflake) and cursor.lastrowid (the Snowflake Python connector returns None for this attribute because Snowflake has no native rowid concept — see snowflake-connector-python#1201). With neither mechanism available, the ORM receives None as the primary key and raises:

sqlalchemy.orm.exc.FlushError: Instance <MyModel at 0x...> has a NULL identity key after a flush ...

Example that fails:

from sqlalchemy import Column, Identity, Integer, String
from sqlalchemy.orm import declarative_base, Session
Base = declarative_base()
class MyModel(Base):
    __tablename__ = "my_model"
    id = Column(Integer, Identity(start=1, increment=1), primary_key=True)  # does not work with ORM
    name = Column(String)
Base.metadata.create_all(engine)
with Session(engine) as session:
    session.add(MyModel(name="test"))
    session.commit()  # raises FlushError: NULL identity key

The dialect emits a SnowflakeWarning at DDL compile time when Identity() is detected on a primary key column to surface this problem early. The warning is emitted once per unique (table, column) pair per Python process — repeated DDL compilation of the same schema does not produce duplicate output.

To silence the warning entirely, use Python's standard warning filter:

import warnings
from snowflake.sqlalchemy.exc import SnowflakeWarning
warnings.filterwarnings("ignore", category=SnowflakeWarning)

Or set it via the PYTHONWARNINGS environment variable before starting your application:

PYTHONWARNINGS=ignore::snowflake.sqlalchemy.exc.SnowflakeWarning python my_app.py

Workaround — use Sequence() instead:

from sqlalchemy import Column, Integer, Sequence, String
from sqlalchemy.orm import declarative_base, Session
Base = declarative_base()
class MyModel(Base):
    __tablename__ = "my_model"
    id = Column(Integer, Sequence("my_model_id_seq"), primary_key=True)
    name = Column(String)
Base.metadata.create_all(engine)
with Session(engine) as session:
    session.add(MyModel(name="test"))
    session.commit()  # id is populated correctly

Sequence objects are fully supported by the Snowflake dialect and are the recommended way to generate auto-incrementing primary keys when using the ORM. See the Auto-increment Behavior section for more details.

`PUT` with a file stream (`BytesIO`)

The Snowflake connector's PUT command accepts an in-memory file_stream keyword, but SQLAlchemy's execution API treats every keyword passed to Connection.execute() as a bound SQL parameter. Passing file_stream through SQLAlchemy therefore fails (see #337):

snowflake.connector.errors.ProgrammingError: 255001: Binding data in type (bytesio) is not supported.

For example, the intuitive call below does not work:

from io import BytesIO
from sqlalchemy import text
buf = BytesIO()
# ❌ Fails — SQLAlchemy binds file_stream as a SQL parameter:
connection.execute(text("PUT file://example.txt @my_stage"), {"file_stream": buf})
# snowflake.connector.errors.ProgrammingError: 255001:
#   Binding data in type (bytesio) is not supported.

Workaround — use the raw connector cursor for PUT/GET file-stream operations. You can obtain one from the SQLAlchemy engine without opening a second connection:

from io import BytesIO
buf = BytesIO()
df.to_json(buf, orient="records", date_format="iso")
buf.seek(0)
raw = engine.raw_connection()
try:
    cur = raw.cursor()
    cur.execute(
        f"PUT file://example.txt @{stage} auto_compress=true",
        file_stream=buf,
    )
finally:
    raw.close()

Regular SQL (including COPY INTO) continues to work through SQLAlchemy as usual; only the file-stream keyword of PUT/GET requires the raw cursor.

Case-sensitive identifiers

Snowflake stores unquoted identifiers in UPPERCASE and treats them case-insensitively. SQLAlchemy uses lowercase for case-insensitive identifiers. The dialect bridges this gap via normalize_name / denormalize_name, but a few edge cases require explicit opt-in.

How reflection maps Snowflake names to SQLAlchemy names

When the dialect reflects a table, each column name passes through normalize_name, which produces one of three outcomes depending on how the identifier was stored in Snowflake:

Snowflake stored form	How it was created	`normalize_name` returns (default)	`normalize_name` returns (`case_sensitive_identifiers=True`)	SQLAlchemy treats it as
`MYCOL` (all-uppercase)	`CREATE TABLE t (MYCOL INT)` — unquoted	`"mycol"` (plain `str`)	`"mycol"` (plain `str`)	case-insensitive
`mycol` (lowercase)	`CREATE TABLE t ("mycol" INT)` — quoted	`quoted_name("mycol", True)`	`quoted_name("mycol", True)`	case-sensitive
`MyCol` (mixed-case)	`CREATE TABLE t ("MyCol" INT)` — quoted	`"MyCol"` (plain `str`)	`quoted_name("MyCol", True)`	case-sensitive — emitted SQL is `"MyCol"` in both modes (`_requires_quotes` forces quoting for any uppercase character)

With the flag off, the only observable difference for mixed-case names is the Python type: isinstance(name, quoted_name) is False and .quote is None. Emitted SQL, dict key equality, and hashing are identical because quoted_name is a str subclass. Enable the flag when downstream code (e.g. an Alembic render_item hook or a custom preparer subclass) needs to distinguish case-sensitive reflected names by type.

You can observe this directly via the inspector:

from sqlalchemy import inspect
from sqlalchemy.sql.elements import quoted_name
inspector = inspect(engine)
for col in inspector.get_columns("my_table"):
    name = col["name"]
    if isinstance(name, quoted_name) and name.quote:
        print(f"{name!r} — case-sensitive (was created quoted in Snowflake)")
    else:
        print(f"{name!r} — case-insensitive (Snowflake stores as {name.upper()})")

This means that after reflection, accessing a case-insensitive column requires the lowercase name. Given a table created as:

CREATE TABLE my_table (mycol INT);   -- unquoted: stored as MYCOL

metadata.reflect(bind=engine)
t = metadata.tables["my_table"]
t.c.mycol      # correct — Snowflake stored MYCOL, reflected as "mycol"
t.c["MYCOL"]   # KeyError — the reflected key is lowercase

And a case-sensitive column (quoted in Snowflake) is accessed by its exact reflected name:

t.c[quoted_name("mycol", True)]  # correct — matches the reflected quoted_name key
t.c.mycol                        # also works — quoted_name.__eq__ compares by value

Lowercase column names in CLUSTER BY

Column objects wrapped in quoted_name("mycol", True) are treated as case-sensitive by SQLAlchemy. Pass them directly to snowflake_clusterby:

from sqlalchemy.sql.elements import quoted_name
from sqlalchemy import Column, Integer, MetaData, Table
t = Table(
    "my_table",
    MetaData(),
    Column(quoted_name("mycol", True), Integer),
    snowflake_clusterby=[quoted_name("mycol", True)],
)
# Generates: CLUSTER BY ("mycol")

Without quoted_name(..., True) the column name is treated as case-insensitive and Snowflake resolves it as MYCOL.

ALL-UPPERCASE identifiers that are SQL reserved words (e.g. `TABLE`, `SELECT`)

Most identifiers are handled correctly without any flag at the SQL layer:

A quoted lowercase name ("mycol" in Snowflake) reflects as quoted_name("mycol", True) — already case-sensitive, in both flag modes.
A quoted mixed-case name ("MyCol") reflects as a plain str "MyCol" by default; with case_sensitive_identifiers=True it reflects as quoted_name("MyCol", True). Emitted SQL is "MyCol" either way because the preparer's _requires_quotes heuristic force-quotes any name containing uppercase characters, so only code that inspects .quote or isinstance(..., quoted_name) observes a difference.
An unquoted name (MYCOL stored as all-uppercase) reflects as "mycol" — correctly case-insensitive.

The one gap is an identifier whose all-uppercase form is also a SQL reserved word. For example, a table literally named TABLE (created as CREATE TABLE "TABLE" ...) stores as TABLE in Snowflake. When the dialect reflects it, normalize_name("TABLE") cannot tell whether TABLE is a plain uppercase column or the keyword TABLE, so by default it returns "TABLE" unchanged — an all-uppercase string that SQLAlchemy treats as case-sensitive. This causes a key mismatch: the table was reflected under the key "TABLE" but the same normalize_name call made during DDL would produce "table".

Enable case_sensitive_identifiers to fix this: the dialect will return quoted_name("table", True) for any all-uppercase identifier that is a reserved word, matching the standard SQLAlchemy convention:

from sqlalchemy import create_engine
engine = create_engine(
    "snowflake://user:pass@account/db",
    case_sensitive_identifiers=True,
)

Or via URL:

snowflake://user:pass@account/db?case_sensitive_identifiers=True

Hard limit: Enabling this flag changes the dict key used by normalize_name("TABLE") from "TABLE" to quoted_name("table", True). Because hash("TABLE") != hash("table"), any existing code that accesses metadata.tables["TABLE"] by the uppercase key will miss after the flag is enabled. Enabling the flag on an existing codebase requires auditing all string-keyed lookups into metadata.tables and table.c for names that happen to be reserved words.

Case-sensitive schema names — table and model definitions

The most reliable way to declare a case-sensitive schema across multiple tables is MetaData(schema=quoted_name(..., True)). All tables and ORM models that share that MetaData object inherit the schema automatically — no repetition per table:

from sqlalchemy import Column, Integer, MetaData, String, Table
from sqlalchemy.sql.elements import quoted_name
# Declare once — every Table attached to this MetaData inherits it.
metadata = MetaData(schema=quoted_name("myschema", True))
orders = Table("orders", metadata, Column("id", Integer, primary_key=True))
items  = Table("items",  metadata, Column("id", Integer, primary_key=True))
# SQL: "myschema".ORDERS and "myschema".ITEMS
# (table names are plain strings — Snowflake uppercases them)

To also make the table names case-sensitive, wrap them in quoted_name as well:

from sqlalchemy.sql.elements import quoted_name
orders = Table(quoted_name("orders", True), metadata, Column("id", Integer, primary_key=True))
items  = Table(quoted_name("items",  True), metadata, Column("id", Integer, primary_key=True))
# SQL: "myschema"."orders" and "myschema"."items"

When tables belong to different schemas, or when the MetaData is shared and you need per-table control, pass schema explicitly on the first table and reuse its .schema attribute on the rest:

orders = Table("orders", metadata,
               Column("id", Integer, primary_key=True),
               schema=quoted_name("myschema", True))
# Reuse the schema object directly — no need to repeat quoted_name.
items = Table("items", metadata,
              Column("id", Integer, primary_key=True),
              schema=orders.schema)

Table.schema is always a quoted_name instance. Assigning from another table's .schema carries the quote=True flag intact.

For ORM declarative models, set the schema on Base.metadata so every model inherits it:

from sqlalchemy import Column, Integer
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.sql.elements import quoted_name
class Base(DeclarativeBase):
    metadata = MetaData(schema=quoted_name("myschema", True))
class Orders(Base):
    __tablename__ = "orders"
    id = Column(Integer, primary_key=True)
class Items(Base):
    __tablename__ = "items"
    id = Column(Integer, primary_key=True)

For per-model override, use __table_args__:

class Orders(Base):
    __tablename__ = "orders"
    __table_args__ = {"schema": quoted_name("myschema", True)}
    id = Column(Integer, primary_key=True)
class Items(Base):
    __tablename__ = "items"
    __table_args__ = {"schema": Orders.__table__.schema}  # reuse — no repetition
    id = Column(Integer, primary_key=True)

Plain strings with inner double-quotes — '"myschema"' — are also accepted and produce the same result as quoted_name("myschema", True) when the dialect is built with case_sensitive_identifiers=True:

engine = create_engine(
    "snowflake://user:pass@account/db",
    case_sensitive_identifiers=True,
)
Table("orders", metadata, schema='"myschema"')         # → "myschema".ORDERS
Table("orders", metadata, schema='"mydb"."myschema"')  # → "mydb"."myschema".ORDERS

Without the flag, the schema parser extracts the inner value but leaves the quote attribute at its default (None), so the preparer's _requires_quotes heuristic decides whether to re-quote it — and for an all-lowercase value like myschema the heuristic returns False, stripping the quotes from the emitted SQL:

# Flag off (default):
Table("orders", metadata, schema='"myschema"')
# SQL: myschema.ORDERS   ← Snowflake resolves this to MYSCHEMA (case-insensitive)

When case-sensitivity matters, either enable the flag or use quoted_name("myschema", True) / MetaData(schema=quoted_name(..., True)) explicitly — those forms carry quote=True on the value itself and are honoured regardless of the flag. Prefer quoted_name in code that is read often; prefer the quoted-string form together with the flag when the value comes from a config that already embeds the quotes.

Identifiers that contain a literal double-quote character use the standard SQL escape "". The parser handles this correctly — '"my""schema"' is treated as a single identifier named my"schema and round-trips to the SQL form "my""schema" in both flag modes (the embedded " is not a legal identifier character, so _requires_quotes forces quoting regardless of the flag):

Table("orders", metadata, schema='"my""schema"')   # schema name: my"schema
# SQL: "my""schema".ORDERS

Case-sensitive schema names — engine connection

create_snowflake_engine sets the default schema for the connection (equivalent to USE SCHEMA on connect). Use it when all queries in the session target the same case-sensitive schema and you do not want to qualify every table individually:

from snowflake.sqlalchemy import create_snowflake_engine
engine = create_snowflake_engine(
    "snowflake://user:pass@account/mydb",
    schema="myschema",
    case_sensitive_schema=True,
)

When the URL is stored outside Python code (environment variable, alembic.ini, Docker/Kubernetes config), create_snowflake_engine is not available and the schema must be percent-encoded directly in the URL string. Wrap the schema name in %22 (the percent-encoded form of "):

snowflake://user:pass@account/mydb/%22myschema%22

The dialect decodes %22 back to a literal " before passing the value to the Snowflake connector, which then executes USE SCHEMA "myschema" preserving case.

When building the URL programmatically with SQLAlchemy's URL() helper, you can pass the schema with embedded double-quotes directly — the helper's percent-encoding handles them automatically:

from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine
engine = create_engine(URL(
    account="abc123",
    user="testuser1",
    password="secret",
    database="mydb",
    schema='"myschema"',   # literal " characters; URL() encodes them as %22 internally
))

This is equivalent to the %22 string form but avoids manual percent-encoding.

Alembic — hand-written migrations

quoted_name works directly in op.create_table and op.add_column, so hand-written migration files need no special handling:

import sqlalchemy as sa
from alembic import op
from sqlalchemy.sql.elements import quoted_name
def upgrade():
    op.create_table(
        "my_table",
        sa.Column(quoted_name("mycol", True), sa.String()),  # quoted → "mycol" in Snowflake
        sa.Column("name", sa.String()),                      # unquoted → NAME in Snowflake
    )
    op.add_column(
        "my_table",
        sa.Column(quoted_name("extra", True), sa.Integer()),
    )

Alembic — case-sensitive schema configuration

When the Alembic version table or the migrations themselves target a case-sensitive schema, use the %22 form in every place Alembic receives a schema string, because these values are passed as plain strings and create_snowflake_engine is not available at that point:

# alembic/env.py
from sqlalchemy import create_engine
url = "snowflake://user:pass@account/mydb/%22myschema%22"
context.configure(
    url=url,
    target_metadata=target_metadata,
    version_table_schema="%22myschema%22",  # keeps alembic_version table in the same schema
)

When include_schemas=True is enabled, Alembic calls inspector.get_schema_names() and passes each result to the include_name filter. The dialect returns case-sensitive schema names as quoted_name objects (e.g. quoted_name("myschema", True)). Because quoted_name inherits from str, a plain string comparison is safe:

# alembic/env.py
from sqlalchemy.sql.elements import quoted_name
def include_name(name, type_, parent_names):
    if type_ == "schema":
        # quoted_name("myschema", True) == "myschema" → True (str subclass equality)
        return name in {'"myschema"', "myschema"}
    return True
context.configure(
    url="snowflake://user:pass@account/mydb/%22myschema%22",
    target_metadata=target_metadata,
    include_schemas=True,
    include_name=include_name,
    version_table_schema="%22myschema%22",
)

Note: inspector.get_schema_names() returns the schema as Snowflake stores it — lowercase "myschema" if it was created quoted, or uppercase "MYSCHEMA" if unquoted. After normalize_name this becomes quoted_name("myschema", True) or "myschema" (plain lowercase) respectively. Filter against the lowercase value and both cases are covered by the str equality of quoted_name.

Alembic — autogenerate and case-sensitive columns

Alembic's default renderer serialises quoted_name("mycol", True) as the plain string "mycol", losing the case-sensitivity signal. The generated migration would create a case-insensitive MYCOL column instead of "mycol".

This also affects the comparison phase: when autogenerate detects that a reflected quoted_name("mycol", True) column differs from what it would render, it may emit a spurious alter_column on every run. The fix for both problems is the same — register the render_item hook in env.py:

from snowflake.sqlalchemy.alembic_util import render_item as snowflake_render_item
context.configure(
    ...,
    render_item=snowflake_render_item,
)

Hard limit: Alembic has no dialect-level rendering hook. The render_item callback in env.py is the only injection point and requires a two-line opt-in per project. This cannot be eliminated without upstream Alembic changes.

Patterns that do not work

Plain unquoted string as a case-sensitive schema

# WRONG — "myschema" (no quotes, no quoted_name) is denormalized to MYSCHEMA
Table("my_table", metadata, schema="myschema")
class MyModel(Base):
    __tablename__ = "my_table"
    __table_args__ = {"schema": "myschema"}  # also wrong

A plain lowercase string has no quote attribute, so denormalize_name converts it to MYSCHEMA before it reaches Snowflake. No error is raised — if a schema named MYSCHEMA happens to exist the query succeeds silently against the wrong schema.

Use quoted_name("myschema", True), '"myschema"', or MetaData(schema=quoted_name(..., True)) as shown in the section above.

Prerequisites

Snowflake Connector for Python

Data Analytics and Web Application Frameworks (Optional)

Installing Snowflake SQLAlchemy

Verifying Your Installation

Parameters and Behavior

Connection Parameters

Escaping Special Characters such as %, @ signs in Passwords

Using a proxy server

Using session parameters

Sensitive connection parameters

Temporary compatibility: legacy_url_params

Opening and Closing Connection

Automatic reconnection on expired sessions

Auto-increment Behavior

Object Name Case Handling

Index Support

Single Column Index

Multi-Column Index

Numpy Data Type Support

DECFLOAT Data Type Support

DECFLOAT Precision

Returning float Instead of Decimal

VECTOR Data Type Support

UUID Data Type Support

Timestamp and Timezone Support

Cache Column Metadata

Single-Table vs Multi-Table Reflection Performance

Cross-Database Reflection

Reflecting Large Schemas (10,000+ objects)

How paging works

Known limitations

VARIANT, ARRAY and OBJECT Support

Writing Python dicts/lists to VARIANT, OBJECT and ARRAY

Automatic JSON handling with enable_structured_type_json

Reading keys and elements with subscript access

Customizing JSON serialization (json_serializer / json_deserializer)

Structured Data Types Support

MAP

OBJECT

ARRAY

CLUSTER BY Support

ORM / Declarative usage

Row Access Policy Support

RELY Constraint Property

Alembic Support

Key Pair Authentication Support

PrivateLink and Account Name Handling

SSO Authentication (Okta, Azure AD, etc.)

Boolean Column Comparisons (IS TRUE / IS FALSE)

Merge Command Support

Using a VALUES source (no staging table required)

Multi-Table Insert (INSERT ALL / INSERT FIRST)

Bulk Insert Optimization for ORM Models

Bulk Array Binding for Large executemany Inserts

CopyIntoStorage Support

Securing cloud credentials in logs

Creating a named file format

Iceberg Table with Snowflake Catalog support

Hybrid Table support

Dynamic Tables support

Notes

Verifying Package Signatures

Support

Known Limitations

Identity columns as primary keys

PUT with a file stream (BytesIO)

Case-sensitive identifiers

How reflection maps Snowflake names to SQLAlchemy names

Lowercase column names in CLUSTER BY

ALL-UPPERCASE identifiers that are SQL reserved words (e.g. TABLE, SELECT)

Case-sensitive schema names — table and model definitions

Case-sensitive schema names — engine connection

Alembic — hand-written migrations

Alembic — case-sensitive schema configuration

Alembic — autogenerate and case-sensitive columns

Patterns that do not work

Escaping Special Characters such as `%, @` signs in Passwords

Temporary compatibility: `legacy_url_params`

Returning `float` Instead of `Decimal`

Automatic JSON handling with `enable_structured_type_json`

Customizing JSON serialization (`json_serializer` / `json_deserializer`)

Boolean Column Comparisons (`IS TRUE` / `IS FALSE`)

Using a `VALUES` source (no staging table required)

Multi-Table Insert (`INSERT ALL` / `INSERT FIRST`)

Bulk Array Binding for Large `executemany` Inserts

`PUT` with a file stream (`BytesIO`)

ALL-UPPERCASE identifiers that are SQL reserved words (e.g. `TABLE`, `SELECT`)