DOCSEARCHbyAlgolia
Submit Docs

Snowflake SQLAlchemy Dialect

repository·main·Indexed 42 days ago

https://github.com/snowflakedb/snowflake-sqlalchemy

A SQLAlchemy dialect that bridges Snowflake databases with SQLAlchemy applications, built on top of the Snowflake Connector for Python. It supports asynchronous database access via SQLAlchemy's asyncio extension (requiring Python 3.11+, snowflake-connector-python 5.x, and SQLAlchemy 2.0.44+), Hybrid Table indexing, and specialized data type handling for NumPy and DECFLOAT.

Tokens
26K
Snippets
81
Records
92
Agent score
54%

What's inside Snowflake SQLAlchemy

  1. Handle object name casing in Snowflake SQLAlchemy

    main

    Snowflake stores case-insensitive object names in uppercase. SQLAlchemy treats lowercase names as case-insensitive.

    To avoid mismatches during table and index reflection, use all lowercase names on the SQLAlchemy side unless the identifiers were explicitly created as case-sensitive using quotes (e.g., "TestDb"). If you use uppercase names, SQLAlchemy will enclose them in quotes, which may cause mismatches against the Snowflake data dictionary.

  2. Avoid syntax errors with Boolean Column Comparisons

    main

    Snowflake does not support the SQL syntax IS TRUE or IS FALSE. Using SQLAlchemy's idiomatic col.is_(True) or col.is_(False) will raise a ProgrammingError.

    Correct Approach: Use equality against SQLAlchemy's true() or false() helpers.

    Note on NULL semantics: flag == true() will filter out NULL rows (returning NULL for the expression). If you require strict IS TRUE behavior (where NULL is treated as FALSE), you must explicitly check for non-null values using and_ and isnot(None).

    from sqlalchemy import true, false, select, and_

# ✅ Correct: Use equality with true()/false()
select(t).where(t.c.flag == true())

# ✅ Correct: Strict IS TRUE behavior (handles NULLs)
select(t).where(and_(t.c.flag.isnot(None), t.c.flag == true()))

# ❌ Incorrect: Raises syntax error in Snowflake
select(t).where(t.c.flag.is_(True))

  3. Handle Boolean comparisons in Snowflake

    main

    Snowflake does not support the IS TRUE or IS FALSE syntax. When using SQLAlchemy, you must use explicit equality comparisons with true() or false() functions.

    Incorrect: col.is_(True) (may fail depending on context) Correct: col == true() or col == false()

    from sqlalchemy import true, false

# Use these patterns for boolean columns
query = select(my_table).where(my_table.c.is_active == true())

  4. Configure Auto-incrementing columns

    main

    Snowflake SQLAlchemy supports two ways to handle auto-incrementing values:

    1. Sequence (Recommended for ORM): Uses a Sequence object. SQLAlchemy fetches the next value via a separate SELECT <seq>.nextval before the INSERT. This ensures the value is available to the ORM immediately after insert.
    2. AUTOINCREMENT: Uses SQLAlchemy's autoincrement=True. Snowflake assigns the value server-side during a single INSERT statement. This is faster but the value is not returned to the ORM in the same step.

    Note: For Hybrid tables, both methods work, but Sequence is preferred for ORM-mapped classes.

    from sqlalchemy import Table, Column, Integer, String, MetaData, Sequence

metadata = MetaData()

# Using Sequence (Recommended for ORM)
t = Table('mytable', metadata,
    Column('id', Integer, Sequence('id_seq'), primary_key=True),
    Column('name', String),
)

# Using AUTOINCREMENT (Single-statement insert)
t_auto = Table('mytable_auto', metadata,
    Column('id', Integer, primary_key=True, autoincrement=True),
    Column('name', String),
)

  5. Optimize table inspection and reflection performance

    main

    Snowflake SQLAlchemy uses different query strategies depending on whether you are inspecting a single table or reflecting an entire schema.

    • Single-table inspection: Using Inspector methods like get_columns(), get_pk_constraint(), get_unique_constraints(), get_foreign_keys(), or get_indexes() triggers per-table queries (DESC TABLE or SHOW ... IN TABLE). This is efficient for looking up specific tables.
    • Bulk reflection: Using MetaData.reflect() issues schema-wide queries (e.g., SHOW PRIMARY KEYS IN SCHEMA). This is significantly faster for schemas with many tables (100+) because it avoids thousands of round-trips.

    Best Practices:

    1. Use metadata.reflect() for bulk operations to leverage cached schema-wide queries.
    2. Use inspector.get_columns() etc. for targeted single-table lookups.
    3. For very large schemas, limit the scope of reflection using the only parameter to avoid unnecessary overhead.
    # For bulk reflection of specific tables only
metadata.reflect(bind=engine, schema='public', only=['table1', 'table2'])

# For single-table inspection
inspector = inspect(engine)
columns = inspector.get_columns('my_table', schema='public')

  6. Optimize schema reflection performance

    main

    Snowflake SQLAlchemy optimizes metadata inspection to avoid slow per-table queries in large schemas.

    • Bulk Reflection (Fast): Using MetaData.reflect(bind=engine, schema='...') or Table(..., autoload_with=engine) in SQLAlchemy 2.x triggers schema-wide SHOW or information_schema queries. This caches results for all tables in the schema at once.
    • Single-Table Inspection (Direct): Using inspector.get_columns(table_name) issues a direct DESC TABLE query. This is faster for specific tables or temporary tables.

    Note on reflected types: Because inspector.get_columns() uses DESC TABLE, reflected types will show Snowflake's resolved default sizes (e.g., VARCHAR(16777216)). Use isinstance() for type checks rather than string comparisons.

    from sqlalchemy import MetaData, inspect, create_engine

engine = create_engine('snowflake://...')

# Bulk reflection: one schema-wide query per metadata type, all tables cached
metadata = MetaData()
metadata.reflect(bind=engine, schema='public')

  7. Enable structured JSON handling with enable_structured_type_json

    main

    The enable_structured_type_json dialect flag allows for automatic deserialization of semi-structured columns (VARIANT, OBJECT, and ARRAY) into native Python types (dict, list, etc.) when reading, and automatic serialization when writing.

    Behavior:

    • Reading: Deserializes JSON text from Snowflake into native Python objects.
    • Writing: Serializes native Python objects and wraps them in PARSE_JSON. Because Snowflake rejects functions in a VALUES clause, writes are rendered as INSERT ... SELECT (using UNION ALL for multi-row inserts) or as SET col = PARSE_JSON(...) for UPDATE statements.
    • Typed Columns: Columns explicitly defined as OBJECT(...), ARRAY(<type>), or MAP continue to use native connector handling.

    Configuration:

    • This flag is off by default in version v2.0.0a2.
    • It can be enabled via a URL parameter or by passing enable_structured_type_json=True to create_engine().
    • When enabled, the engine's json_serializer and json_deserializer parameters are used if provided.
    # Enabling via create_engine
engine = create_engine(
    "snowflake://...",
    enable_structured_type_json=True,
    json_serializer=my_serializer,
    json_deserializer=my_deserializer
)

  8. Handle large Snowflake schemas (10,000+ objects)

    main

    Snowflake SHOW commands are limited to 10,000 rows per call. Snowflake SQLAlchemy automatically handles this by transparently paging through object-listing commands (tables, views, sequences, schemas, etc.) using LIMIT <n> FROM '<last_name>' logic.

    No manual code changes are required for the following Inspector methods to work on schemas exceeding 10,000 objects:

    • get_table_names()
    • get_view_names()
    • get_sequence_names()
    • get_schema_names()
    • get_temp_table_names()

    Limitation: Schema-wide constraint/index commands (e.g., SHOW PRIMARY KEYS IN SCHEMA) are not yet paged and may still hit the 10,000-row cap during MetaData.reflect(). However, single-table inspection via Inspector remains unaffected.

  9. Async usage notes and limitations

    main

    When using the async dialect, follow these standard SQLAlchemy asyncio patterns:

    • No lazy loading: Accessing unloaded ORM relationships outside an active greenlet context will raise an error. Use eager loading (selectinload(), joinedload()), the AsyncAttrs mixin, or AsyncSession.run_sync().
    • Session configuration: It is common to set expire_on_commit=False on async_sessionmaker() to prevent attribute-expiry-triggered lazy loads after a commit.
    • Engine lifecycle: Dispose engines using await engine.dispose() if you are moving between unrelated event loops.
    • Alembic: There is currently no async-aware Alembic migration runner; use a synchronous engine/connection for migrations.
    • Connector status: The underlying connector dependency is a pre-release (5.0.0rc3), so expect potential breaking changes.

  10. Install Snowflake SQLAlchemy

    main

    Install the snowflake-sqlalchemy package from PyPI using pip. This will automatically install the required snowflake-connector-python dependency.

    Note: If you are using data analytics frameworks like Pandas or Jupyter, using an Anaconda environment is recommended to avoid manual compilation of C tools.

    pip install --upgrade snowflake-sqlalchemy

  11. Reflect tables across different databases

    main

    You can reflect tables located in different databases within a single SQLAlchemy session by using the database.schema notation in the schema parameter. This allows for cross-database joins without writing raw SQL.

    Note on special characters: If a schema name contains a literal dot (e.g., my.schema), you must wrap the schema part in double quotes within the string: database."my.schema".

    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
)

  12. Migrate from Snowflake SQLAlchemy 1.x to 2.x

    main

    To upgrade snowflake-sqlalchemy from the 1.x line (up to 1.11.0) to the 2.x line (e.g., 2.0.0a2), follow these steps:

    1. Upgrade SQLAlchemy: Ensure your application is compatible with SQLAlchemy 2.0. If you are currently on SQLAlchemy 1.4, migrate your application code to SQLAlchemy 2.0 first.
    2. Upgrade Python: Ensure you are using Python 3.11 through 3.14.
    3. Upgrade the Dialect: Since 2.x is currently in pre-release, you must use the --pre flag or pin the specific version.
    4. Verify: Check the version after installation.
    5. Test: Run your test suite, paying close attention to reflection-dependent code and regexp_* usage.
    # Install a specific 2.x preview
pip install "snowflake-sqlalchemy==2.0.0a2"

# Or upgrade to the newest available 2.x pre-release
pip install --upgrade --pre "snowflake-sqlalchemy>=2.0.0"

# Verify the installation
python -c "import snowflake.sqlalchemy; print(snowflake.sqlalchemy.__version__)"