### Set Up Snowflake SQLAlchemy Test Environment

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/tests/README.rst

This bash script prepares a virtual environment for testing Snowflake SQLAlchemy. It installs pytest, numpy, pandas, and the built wheel package. It also includes instructions to configure connection parameters in 'tests/parameters.py'.

```bash
pyvenv /tmp/test_snowflake_sqlalchemy
source /tmp/test_snowflake_sqlalchemy/bin/activate
pip install Cython
pip install pytest numpy pandas
pip install dist/snowflake_sqlalchemy*.whl
vim tests/parameters.py
```

--------------------------------

### Alembic Snowflake Migration Script Example

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

An example of an Alembic migration script for creating a 'users' table in Snowflake. It demonstrates the use of `op.create_table` and includes the custom `VARIANT` type from `snowflake.sqlalchemy` for metadata storage.

```python
"""Add users table

Revision ID: abc123
"""
from alembic import op
import sqlalchemy as sa
from snowflake.sqlalchemy import VARIANT

def upgrade():
    op.create_table(
        'users',
        sa.Column('id', sa.Integer(), primary_key=True),
        sa.Column('name', sa.String(100), nullable=False),
        sa.Column('metadata', VARIANT()),
    )

def downgrade():
    op.drop_table('users')
```

--------------------------------

### Construct Snowflake Connection Strings

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Examples of connection string formats for Snowflake SQLAlchemy, including optional parameters for database, schema, warehouse, and role.

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

--------------------------------

### Establish Snowflake Engine Connection

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates how to create a SQLAlchemy engine for Snowflake using both standard connection strings and the snowflake.sqlalchemy.URL helper method. Includes examples for passing additional connection arguments like timezones.

```python
from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL

# Using connection string
engine = create_engine('snowflake://testuser1:0123456@abc123/testdb/public?warehouse=testwh&role=myrole')

# Using URL helper
engine = create_engine(URL(
    account = 'abc123',
    user = 'testuser1',
    password = '0123456',
    database = 'testdb',
    schema = 'public',
    warehouse = 'testwh',
    role='myrole',
    timezone = 'America/Los_Angeles',
))
```

--------------------------------

### Verify Snowflake SQLAlchemy Installation

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

A Python script to test the connection to a Snowflake instance by executing a version query. It requires the sqlalchemy library and valid Snowflake credentials.

```python
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()
```

--------------------------------

### Build Snowflake SQLAlchemy Wheel Package

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/tests/README.rst

This bash script clones the Snowflake SQLAlchemy repository, sets up a virtual environment, installs necessary build tools, and creates a wheel package for the library. The resulting wheel file can be found in the './dist' directory.

```bash
git clone git@github.com:snowflakedb/snowflake-sqlalchemy.git
cd snowflake-sqlalchemy
pyvenv /tmp/test_snowflake_sqlalchemy
source /tmp/test_snowflake_sqlalchemy/bin/activate
pip install -U pip setuptools wheel
python setup.py bdist_wheel
```

--------------------------------

### Define Iceberg Table with ARRAY Type in Python

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Provides an example of creating an Iceberg table with an ARRAY column using Snowflake SQLAlchemy. This snippet illustrates how to specify the element type and nullability for the array.

```python
from snowflake.sqlalchemy import MAP, OBJECT, ARRAY
from sqlalchemy import Table, Column, Integer, String, MetaData, text, TEXT

# Assuming 'metadata', 'engine', 'table_name', 'external_volume', 'base_location' are defined
# Example for ARRAY
array_table = Table(
    'iceberg_array_table',
    metadata,
    Column("id", Integer, primary_key=True),
    Column("array_col", ARRAY(TEXT(16777216))),
    # Add other columns as needed
    # For actual Iceberg table creation, you'd use a specific IcebergTable construct if available
)
# metadata.create_all(engine)
```

--------------------------------

### Snowflake Connection and Transaction Management

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Provides examples of managing database connections and transactions with Snowflake using SQLAlchemy. It emphasizes the use of context managers (`with engine.connect()`) for automatic resource cleanup and demonstrates explicit transaction control with `commit()` and `rollback()`.

```python
from sqlalchemy import create_engine, text
from snowflake.sqlalchemy import URL

engine = create_engine(URL(
    account='abc123', user='user', password='pass',
    database='db', schema='public', warehouse='wh'
))

# Recommended: Use context manager for automatic cleanup
try:
    with engine.connect() as connection:
        # Execute queries
        result = connection.execute(text("SELECT * FROM users"))
        rows = result.fetchall()

        # Explicit transaction control
        connection.execute(text("INSERT INTO users (name) VALUES ('Bob')"))
        connection.commit()  # Explicit commit

        # Rollback on error
        try:
            connection.execute(text("INSERT INTO users (name) VALUES (NULL)"))
            connection.commit()
        except Exception:
            connection.rollback()
            raise
finally:
    engine.dispose()  # Clean up connection pool
```

--------------------------------

### Define Iceberg Table with OBJECT Type in Python

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Shows the definition of an Iceberg table featuring an OBJECT column with Snowflake SQLAlchemy. This example demonstrates specifying field types and nullability within the OBJECT definition.

```python
from snowflake.sqlalchemy import MAP, OBJECT, ARRAY
from sqlalchemy import Table, Column, Integer, String, MetaData, text, NUMBER, TEXT

# Assuming 'metadata', 'engine', 'table_name', 'external_volume', 'base_location' are defined
# Example for OBJECT
object_table = Table(
    'iceberg_object_table',
    metadata,
    Column("id", Integer, primary_key=True),
    Column(
        "object_col",
        OBJECT(key1=(TEXT(16777216), False), key2=(NUMBER(10, 0), False))
        # Alternatively, without explicit nullable flag if default is acceptable:
        # OBJECT(key1=TEXT(16777216), key2=NUMBER(10, 0))
    ),
    # Add other columns as needed
    # For actual Iceberg table creation, you'd use a specific IcebergTable construct if available
)
# metadata.create_all(engine)
```

--------------------------------

### NumPy Datetime64 Data Round Trip with Snowflake SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Shows how to perform a round trip of `numpy.datetime64` data using Snowflake SQLAlchemy. This requires enabling NumPy support by setting `numpy=True` in the connection parameters. The example uses pandas for reading data back from Snowflake.

```python
import numpy as np
import pandas as pd
from sqlalchemy import create_engine
from snowflake.sqlalchemy import URL

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
```

--------------------------------

### Configure Snowflake SQLAlchemy Connection Parameters

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/tests/README.rst

This Python code snippet defines the connection parameters for Snowflake SQLAlchemy in a dictionary format. This is used within the 'tests/parameters.py' file for setting up database connections during testing.

```python
CONNECTION_PARAMETERS = {
    'account':  'testaccount',
    'user':     'user1',
    'password': 'testpasswd',
    'schema':   'testschema',
    'database': 'testdb',
}
```

--------------------------------

### Configure Key Pair Authentication for Snowflake SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates how to load a private key from a file and pass it to the SQLAlchemy engine via connect_args. This allows for secure authentication without storing plain-text passwords.

```python
from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
import os

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})
```

--------------------------------

### Configure Session Parameters

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Shows how to set session-level parameters such as QUERY_TAG using the connect_args dictionary or by executing ALTER SESSION commands mid-session.

```python
from sqlalchemy import create_engine, text

# Set via connect_args
engine = create_engine(
    URL(...),
    connect_args={"session_parameters": {"QUERY_TAG": "SOME_QUERY_TAGS"}}
)

# Set mid-session
with engine.connect() as conn:
    conn.execute(text("ALTER SESSION SET QUERY_TAG = 'batch_job_1'"))
```

--------------------------------

### Create Dynamic Tables with SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Shows how to define Dynamic Tables using Core syntax. Supports both manual column definition with raw SQL queries and automatic column inference using SQLAlchemy select constructs.

```python
dynamic_test_table_1 = DynamicTable(
    "dynamic_MyUser",
    metadata,
    Column("id", Integer),
    Column("name", String),
    target_lag=(1, TimeUnit.HOURS),
    warehouse='test_wh',
    refresh_mode=SnowflakeKeyword.FULL,
    as_query="SELECT id, name from MyUser;"
)
```

```python
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)
)
```

--------------------------------

### Manage Connection Lifecycle

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Best practices for opening and closing connections to ensure resources are released correctly and the Snowflake session is terminated properly.

```python
try:
    with engine.connect() as connection:
        connection.execute(text(<SQL>))
finally:
    engine.dispose()
```

--------------------------------

### Verify Package Signature with Cosign

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Provides the command-line instructions to verify the authenticity of the Snowflake SQLAlchemy Python package using the Sigstore cosign tool.

```bash
./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
```

--------------------------------

### Create Hybrid Tables with SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates how to define Hybrid Tables in Snowflake using both SQLAlchemy Core and declarative patterns. These tables support indexing for optimized performance.

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

```python
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)
```

--------------------------------

### Inspect schema metadata and configure caching

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Illustrates the use of the SQLAlchemy inspection API to retrieve table and column metadata, and demonstrates the historical usage of the cache_column_metadata flag.

```python
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)
```

```python
engine = create_engine(URL(
    account = 'abc123',
    user = 'testuser1',
    password = 'pass',
    database = 'db',
    schema = 'public',
    warehouse = 'testwh',
    role='myrole',
    cache_column_metadata=True,
))
```

--------------------------------

### Enable DECFLOAT Precision in Snowflake SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/tests/README.rst

This Python code demonstrates how to enable full 38-digit precision for Snowflake's DECFLOAT type when using SQLAlchemy. By adding `enable_decfloat=True` to the connection URL, you prevent truncation of large decimal values.

```python
from decimal import Decimal
from sqlalchemy import create_engine, Column, Integer, MetaData, Table, select
from snowflake.sqlalchemy import DECFLOAT

# Use enable_decfloat=True in the connection URL for full precision
engine = create_engine(
    'snowflake://user:password@account/database/schema?enable_decfloat=True'
)
metadata = MetaData()

prices = Table(
    'prices', metadata,
    Column('id', Integer, primary_key=True),
    Column('value', DECFLOAT()),
)
metadata.create_all(engine)

with engine.connect() as conn:
    # Insert a value with 38 significant digits
    conn.execute(prices.insert().values(
        id=1,
        value=Decimal('12345678901234567890123456789.123456789')
    ))
    conn.commit()

    # Query returns full precision with enable_decfloat=True
    result = conn.execute(select(prices)).fetchone()
    print(result[1])
```

--------------------------------

### Alembic Integration for Snowflake Dialect in Python

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Shows the necessary code to add to `alembic/env.py` to enable Alembic, a database migration tool, to recognize and work with the Snowflake SQLAlchemy dialect. This involves creating a custom Alembic implementation.

```python
from alembic.ddl.impl import DefaultImpl

class SnowflakeImpl(DefaultImpl):
    __dialect__ = 'snowflake'

# Ensure this class definition is placed within your alembic/env.py file.
```

--------------------------------

### Manually Set Python Decimal Context Precision

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/tests/README.rst

This Python snippet shows an alternative method to handle large decimal precision by manually setting the global decimal context precision to 38 digits. This can be used when the `enable_decfloat=True` connection parameter is not feasible.

```python
import decimal
decimal.getcontext().prec = 38
```

--------------------------------

### Create Table with CLUSTER BY Clause in Python

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates how to define a Snowflake table with clustering keys using the `snowflake_clusterby` parameter in SQLAlchemy. This allows for performance optimization by co-locating related data.

```python
from sqlalchemy import Table, Column, Integer, String, MetaData, text

# Assuming 'metadata' and 'engine' are already defined
metadata = MetaData()

t = Table('myuser', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String),
    # Example with column names and a text expression for clustering
    snowflake_clusterby=['id', 'name', text('id > 5')])

# metadata.create_all(engine)
```

--------------------------------

### Create Snowflake SQLAlchemy Database Connections

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Demonstrates how to create SQLAlchemy database engines for Snowflake using the URL helper function. It covers basic connections, handling special characters in passwords, and configuring session parameters. Requires the `snowflake-sqlalchemy` and `sqlalchemy` libraries.

```python
from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine, text

# Basic connection with account, user, and password
engine = create_engine(
    URL(
        account='abc123',
        user='testuser1',
        password='0123456',
        database='testdb',
        schema='public',
        warehouse='testwh',
        role='myrole',
    )
)

# Connection with special characters in password (URL-encoded)
import urllib.parse
quoted_password = urllib.parse.quote("kx@% jj5/g")
engine = create_engine(
    URL(
        account='abc123',
        user='testuser1',
        password=quoted_password,
        database='testdb',
        schema='public',
    )
)

# Connection with session parameters
engine = create_engine(
    URL(account='abc123', user='testuser1', password='pass', database='db'),
    connect_args={
        "session_parameters": {
            "QUERY_TAG": "my_application",
        }
    },
)

# Execute queries
with engine.connect() as connection:
    result = connection.execute(text("SELECT current_version()"))
    print(result.fetchone()[0])  # Output: e.g., "8.15.0"
```

--------------------------------

### Execute Python Validation Script

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Command line instruction to execute the verification script created in the previous step.

```shell
python validate.py
```

--------------------------------

### Create Multi-Column Index in Snowflake SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Illustrates the creation of a multi-column index for Hybrid Tables using Snowflake SQLAlchemy. This involves defining an `Index` object and specifying the columns to be included in the index. The code relies on SQLAlchemy and Snowflake SQLAlchemy.

```python
from sqlalchemy import Column, Integer, String, MetaData, Index
from snowflake.sqlalchemy import HybridTable

metadata = MetaData()
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)
```

--------------------------------

### Define Iceberg Tables in Snowflake SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Shows how to define Iceberg tables using either SQLAlchemy Core syntax or a declarative model approach. Requires specifying external volume and base location parameters.

```python
# Core syntax
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"
)

# 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)
```

--------------------------------

### Define and Use Standard Data Types in Snowflake SQLAlchemy

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Illustrates defining Snowflake tables using SQLAlchemy's standard data types and Snowflake-specific aliases. It shows table creation, insertion, and querying. Requires `snowflake-sqlalchemy` and `sqlalchemy`.

```python
from sqlalchemy import Column, Integer, String, MetaData, Table, text
from snowflake.sqlalchemy import (
    BIGINT, BINARY, BOOLEAN, CHAR, DATE, DATETIME, DECIMAL, FLOAT,
    INT, INTEGER, REAL, SMALLINT, TIME, TIMESTAMP, VARCHAR,
    NUMBER, STRING, TEXT, DOUBLE, FIXED, DEC, TINYINT, BYTEINT
)

# Assume 'engine' is already created as shown in the connection example
# engine = create_engine(URL(...))

metadata = MetaData()

# Create table with standard types
users = Table('users', metadata,
    Column('id', INTEGER, primary_key=True),
    Column('name', VARCHAR(100)),
    Column('email', STRING),  # Alias for VARCHAR
    Column('age', SMALLINT),
    Column('balance', NUMBER(10, 2)),  # DECIMAL with precision/scale
    Column('score', DOUBLE),  # Alias for FLOAT
    Column('is_active', BOOLEAN),
    Column('created_at', TIMESTAMP),
    Column('birth_date', DATE),
)

# metadata.create_all(engine) # Uncomment to create the table

# Insert and query
# with engine.connect() as conn:
#     conn.execute(users.insert().values(
#         id=1, name='Alice', email='alice@example.com',
#         age=30, balance=1000.50, score=95.5, is_active=True
#     ))
#     conn.commit()
# 
#     result = conn.execute(users.select())
#     for row in result:
#         print(row)
```

--------------------------------

### Implement Auto-increment with Sequences

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates how to define an auto-incrementing primary key column using the SQLAlchemy Sequence object in a table definition.

```python
from sqlalchemy import Table, Column, Integer, Sequence
t = Table('mytable', metadata,
    Column('id', Integer, Sequence('id_seq'), primary_key=True),
    Column(...)
)
```

--------------------------------

### Create External Stage and File Format in Snowflake

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Demonstrates creating a named CSV file format and an external S3 stage using Snowflake SQLAlchemy. It shows how to define format options like field delimiters and NULL representations, and how to link an S3 bucket with IAM role credentials to the stage. It also illustrates creating sub-paths from an existing stage.

```python
from snowflake.sqlalchemy import (
    CreateStage, CreateFileFormat, ExternalStage,
    AWSBucket, CSVFormatter
)

# Create a named file format
csv_format = CSVFormatter()
csv_format.field_delimiter('|')
csv_format.null_if(['NULL', ''])

create_format = CreateFileFormat(
    format_name='my_csv_format',
    formatter=csv_format,
    replace_if_exists=True
)
connection.execute(create_format)

# Create an external stage pointing to S3
s3_bucket = AWSBucket.from_uri('s3://my-data-bucket/').credentials(
    aws_role='arn:aws:iam::123456789:role/snowflake-role'
)

create_stage = CreateStage(
    container=s3_bucket,
    stage=ExternalStage(
        name='my_s3_stage',
        namespace='my_db.my_schema.',
        file_format='my_csv_format'
    ),
    replace_if_exists=True,
    temporary=False
)
connection.execute(create_stage)

# Use stage in copy operations
stage = ExternalStage(
    name='my_s3_stage',
    path='daily_exports/2024/',
    namespace='my_db.my_schema.'
)

# Create sub-path from existing stage
sub_stage = ExternalStage.from_parent_stage(
    stage,
    path='january',
    file_format='my_csv_format'
)
```

--------------------------------

### Export Data to Cloud Storage using CopyIntoStorage

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates saving table data to external cloud storage (AWS S3) using the CopyIntoStorage expression. It includes support for encryption and custom CSV formatting.

```python
from snowflake.sqlalchemy import CopyIntoStorage, AWSBucket, CSVFormatter

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)
```

--------------------------------

### Configure Dynamic Tables for Automated Refresh

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Demonstrates creating Dynamic Tables that refresh automatically based on source data. Supports both raw SQL queries and SQLAlchemy select statements for defining the refresh logic.

```python
from sqlalchemy import Column, Integer, String, MetaData, select, func
from snowflake.sqlalchemy import DynamicTable, TimeUnit, SnowflakeKeyword

metadata = MetaData()

# Dynamic table with string query
dynamic_summary = DynamicTable(
    'user_summary',
    metadata,
    Column('department', String),
    Column('user_count', Integer),
    target_lag=(1, TimeUnit.HOURS),
    warehouse='compute_wh',
    refresh_mode=SnowflakeKeyword.AUTO,
    as_query="SELECT department, COUNT(*) as user_count FROM users GROUP BY department"
)

# Dynamic table using SQLAlchemy select()
base_table = Table('users', metadata, autoload_with=engine)

dynamic_stats = DynamicTable(
    'user_stats',
    metadata,
    target_lag=SnowflakeKeyword.DOWNSTREAM,
    warehouse='compute_wh',
    as_query=select(
        base_table.c.department,
        func.count().label('total'),
        func.avg(base_table.c.age).label('avg_age')
    ).group_by(base_table.c.department)
)
```

--------------------------------

### Create Single Column Index in Snowflake SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates creating a single column index for Hybrid Tables in Snowflake SQLAlchemy. This can be achieved by setting `index=True` on a column or by explicitly defining an `Index` object. The code requires SQLAlchemy and Snowflake SQLAlchemy libraries.

```python
from sqlalchemy import Column, Integer, String, MetaData, Index
from snowflake.sqlalchemy import HybridTable

metadata = MetaData()
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)
```

--------------------------------

### Implement Snowflake TIMESTAMP types

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Shows how to map Snowflake-specific timestamp types (NTZ, TZ, LTZ) and how generic SQLAlchemy DateTime types handle timezone-aware columns.

```python
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()),
    Column('scheduled_at', TIMESTAMP_TZ()),
    Column('logged_at', TIMESTAMP_LTZ()),
)
metadata.create_all(engine)
```

```python
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()),
    Column('aware_ts', DateTime(timezone=True)),
    Column('naive_ts2', TIMESTAMP()),
    Column('aware_ts2', TIMESTAMP(timezone=True)),
)
metadata.create_all(engine)
```

--------------------------------

### Retrieve and Convert VARIANT, ARRAY, OBJECT Data in Python

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Shows how to fetch data from a table containing VARIANT, ARRAY, and OBJECT columns and convert them into native Python data types using `json.loads`. This is essential for working with semi-structured data retrieved from Snowflake.

```python
import json
from sqlalchemy import select

# Assuming 'engine' and table 't' are already defined
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])
```

--------------------------------

### Create Clustered Tables for Query Optimization

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Shows how to define a table with clustering keys using the snowflake_clusterby parameter. This helps optimize query performance on large datasets by physically organizing data based on specified columns.

```python
from sqlalchemy import Column, Integer, String, MetaData, Table, Sequence, text

metadata = MetaData()

# Table with auto-increment sequence and clustering
users = Table('clustered_users', metadata,
    Column('id', Integer, Sequence('user_id_seq'), primary_key=True),
    Column('region', String(50)),
    Column('created_date', String(10)),
    Column('name', String(100)),
    # Cluster by region and date for optimized queries
    snowflake_clusterby=['region', 'created_date'],
)

metadata.create_all(engine)

# Verify clustering
with engine.connect() as conn:
    result = conn.execute(text("SHOW TABLES LIKE 'CLUSTERED_USERS'"))
    row = result.fetchone()
    print(f"Cluster by: {row._mapping.get('cluster_by')}")
```

--------------------------------

### Create Table with VARIANT, ARRAY, OBJECT Data Types in Python

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates how to define a Snowflake table with VARIANT, ARRAY, and OBJECT columns using SQLAlchemy. These semi-structured types are converted to Python strings and require `json.loads` for conversion to native data types.

```python
from snowflake.sqlalchemy import (VARIANT, ARRAY, OBJECT)
from sqlalchemy import Table, Column, Integer, String, MetaData, text

# Assuming 'metadata' and 'engine' are already defined
metadata = MetaData()

t = Table('my_semi_strucutred_datatype_table', metadata,
    Column('va', VARIANT),
    Column('ob', OBJECT),
    Column('ar', ARRAY))
metadata.create_all(engine)
```

--------------------------------

### Define VECTOR data types in SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates how to define table columns using the Snowflake-specific VECTOR type, specifying the element type and dimension. This requires the snowflake-sqlalchemy library.

```python
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)
```

--------------------------------

### Enable Full DECFLOAT Precision in Snowflake SQLAlchemy Connection

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Shows how to enable full 38-digit precision for `DECFLOAT` values when connecting to Snowflake using SQLAlchemy. This is achieved by adding `enable_decfloat=True` to the connection string or URL parameters. This setting modifies the Python decimal context.

```python
from sqlalchemy import create_engine

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

```python
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,
))
```

--------------------------------

### Snowflake Schema Reflection with SQLAlchemy

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Demonstrates how to use SQLAlchemy's inspection API to reflect database schemas, tables, columns, and constraints in Snowflake. It connects to Snowflake using `snowflake.sqlalchemy.URL` and then uses `inspect` to retrieve schema information.

```python
from sqlalchemy import create_engine, inspect, MetaData, Table
from snowflake.sqlalchemy import URL

engine = create_engine(URL(
    account='abc123', user='user', password='pass',
    database='testdb', schema='public', warehouse='wh'
))

inspector = inspect(engine)

# Get schema information
schemas = inspector.get_schema_names()
print(f"Schemas: {schemas}")

# Get tables in schema
tables = inspector.get_table_names(schema='public')
print(f"Tables: {tables}")

# Get views
views = inspector.get_view_names(schema='public')
print(f"Views: {views}")

# Get detailed column information
for table_name in tables:
    columns = inspector.get_columns(table_name, schema='public')
    pk = inspector.get_pk_constraint(table_name, schema='public')
    fks = inspector.get_foreign_keys(table_name, schema='public')
    unique = inspector.get_unique_constraints(table_name, schema='public')

    print(f"\nTable: {table_name}")
    print(f"  Primary Key: {pk}")
    print(f"  Columns: {[c['name'] for c in columns]}")
    print(f"  Foreign Keys: {fks}")
    print(f"  Unique Constraints: {unique}")

# Reflect existing table
metadata = MetaData()
existing_table = Table('users', metadata, autoload_with=engine)
print(f"Reflected columns: {existing_table.columns.keys()}")
```

--------------------------------

### Define Iceberg Tables for External Storage

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Shows how to define an Iceberg table in Snowflake, which uses the Apache Iceberg format and points to an external storage volume for data persistence.

```python
from sqlalchemy import Column, Integer, String, MetaData
from snowflake.sqlalchemy import IcebergTable

metadata = MetaData()

# Basic Iceberg table
iceberg_events = IcebergTable(
    'iceberg_events',
    metadata,
    Column('event_id', Integer, primary_key=True),
    Column('event_type', String(100)),
    Column('payload', String),
    external_volume='my_s3_volume',
    base_location='events/iceberg',
)
```

--------------------------------

### Define and manipulate semi-structured data with VARIANT, ARRAY, and OBJECT

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Demonstrates how to define tables with semi-structured JSON types and use PARSE_JSON for insertion. It also shows how to retrieve and deserialize these values into Python objects using the json library.

```python
import json
from sqlalchemy import Column, Integer, MetaData, Table, select
from snowflake.sqlalchemy import VARIANT, ARRAY, OBJECT

metadata = MetaData()

# Create table with semi-structured types
semi_structured = Table('semi_structured_data', metadata,
    Column('id', Integer, primary_key=True),
    Column('json_data', VARIANT),      # Any JSON value
    Column('tags', ARRAY),             # JSON array
    Column('metadata', OBJECT),        # JSON object
)

metadata.create_all(engine)

# Insert semi-structured data using PARSE_JSON
with engine.connect() as conn:
    conn.execute(text("""
        INSERT INTO semi_structured_data (id, json_data, tags, metadata)
        SELECT 1,
               PARSE_JSON('{"key": "value", "nested": {"a": 1}}'),
               PARSE_JSON('["tag1", "tag2", "tag3"]'),
               PARSE_JSON('{"created_by": "admin", "version": 2}')
    """))
    conn.commit()

    # Retrieve and parse
    result = conn.execute(select(semi_structured)).fetchone()
    json_data = json.loads(result.json_data)
    tags = json.loads(result.tags)
    metadata_obj = json.loads(result.metadata)
```

--------------------------------

### Snowflake Connection with Key Pair Authentication

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Establishes a secure connection to Snowflake using private key authentication. This method involves loading a private key from a file, converting it to DER format, and passing it as connection arguments to SQLAlchemy's create_engine. This enhances security by avoiding password-based authentication.

```python
from snowflake.sqlalchemy import URL
from sqlalchemy import create_engine
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import serialization
import os

# Load private key from file
with open("rsa_key.p8", "rb") as key_file:
    private_key = serialization.load_pem_private_key(
        key_file.read(),
        password=os.environ['PRIVATE_KEY_PASSPHRASE'].encode(),
        backend=default_backend()
    )

# Convert to DER format for Snowflake connector
private_key_bytes = private_key.private_bytes(
    encoding=serialization.Encoding.DER,
    format=serialization.PrivateFormat.PKCS8,
    encryption_algorithm=serialization.NoEncryption()
)

# Create engine with private key authentication
engine = create_engine(
    URL(
        account='abc123',
        user='testuser1',
        database='testdb',
        schema='public',
        warehouse='testwh',
    ),
    connect_args={
        'private_key': private_key_bytes,
    },
)

with engine.connect() as conn:
    result = conn.execute(text("SELECT CURRENT_USER()"))
    print(result.fetchone()[0])
```

--------------------------------

### Define Iceberg Tables with Snowflake SQLAlchemy

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Demonstrates how to define Iceberg tables with structured types like ARRAY and MAP, as well as creating tables directly from query results. Requires the snowflake-sqlalchemy dialect.

```python
iceberg_users = IcebergTable(
    'iceberg_users',
    metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(100)),
    Column('tags', ARRAY(TEXT(50))),
    Column('attributes', MAP(TEXT(50), TEXT(200))),
    external_volume='my_external_volume',
    base_location='users/iceberg',
)

iceberg_summary = IcebergTable(
    'iceberg_summary',
    metadata,
    Column('category', String),
    Column('total', Integer),
    external_volume='my_external_volume',
    base_location='summary/iceberg',
    as_query="SELECT category, COUNT(*) as total FROM source_table GROUP BY category"
)

metadata.create_all(engine)
```

--------------------------------

### Configure CSV Formatter for Snowflake

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Configures a CSV formatter with various options for Snowflake data loading. This includes setting delimiters, file extensions, date/time formats, binary encoding, escape characters, enclosure options, NULL handling, header skipping, whitespace trimming, and error handling for column count mismatches.

```python
from snowflake.sqlalchemy import CSVFormatter

csv_format = CSVFormatter()
csv_format.compression('gzip')                    # gzip, bz2, brotli, zstd, deflate, raw_deflate, None
csv_format.record_delimiter('\n')                 # Row separator
csv_format.field_delimiter(',')                   # Column separator
csv_format.file_extension('.csv.gz')              # Output file extension
csv_format.date_format('YYYY-MM-DD')              # Date format
csv_format.time_format('HH24:MI:SS')              # Time format
csv_format.timestamp_format('YYYY-MM-DD HH24:MI:SS.FF3')  # Timestamp format
csv_format.binary_format('hex')                   # hex, base64, utf8
csv_format.escape('\\')                           # Escape character
csv_format.escape_unenclosed_field('\\')          # Escape for unenclosed fields
csv_format.field_optionally_enclosed_by('"')      # Quote character: None, ', or "
csv_format.null_if(['NULL', 'null', '', 'N/A'])   # Strings to treat as NULL
csv_format.skip_header(1)                         # Skip header rows on load
csv_format.trim_space(True)                       # Trim whitespace
csv_format.error_on_column_count_mismatch(True)       # Error on column count mismatch
```

--------------------------------

### Encode Special Characters in Passwords

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates how to use urllib.parse to encode special characters in passwords to prevent authentication failures in connection strings.

```python
import urllib.parse
urllib.parse.quote("kx@% jj5/g")
# Output: 'kx%40%25%20jj5/g'

# Example of creating an engine with encoded password
quoted_password = urllib.parse.quote("kx@% jj5/g")
url = f'snowflake://testuser1:{quoted_password}@abc123/testdb/public?warehouse=testwh&role=myrole'
engine = create_engine(url)
```

--------------------------------

### Snowflake ORM Usage with SQLAlchemy Declarative Base

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Illustrates how to use SQLAlchemy's ORM with a declarative base for object-oriented database interactions with Snowflake. It defines `Department` and `Employee` models, including relationships and the `VARIANT` type for metadata, and shows basic session operations.

```python
from sqlalchemy import Column, Integer, String, ForeignKey, create_engine
from sqlalchemy.orm import declarative_base, relationship, sessionmaker
from snowflake.sqlalchemy import URL, VARIANT

Base = declarative_base()

class Department(Base):
    __tablename__ = 'departments'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    employees = relationship("Employee", back_populates="department")

class Employee(Base):
    __tablename__ = 'employees'

    id = Column(Integer, primary_key=True)
    name = Column(String(100), nullable=False)
    email = Column(String(200))
    department_id = Column(Integer, ForeignKey('departments.id'))
    metadata_json = Column(VARIANT)

    department = relationship("Department", back_populates="employees")

# Create engine and tables
engine = create_engine(URL(
    account='abc123', user='user', password='pass',
    database='db', schema='public', warehouse='wh'
))
Base.metadata.create_all(engine)

# Use session for ORM operations
Session = sessionmaker(bind=engine)
session = Session()

# Create records
engineering = Department(name='Engineering')
session.add(engineering)
session.flush()  # Get the ID

alice = Employee(
    name='Alice',
    email='alice@example.com',
    department=engineering
)
session.add(alice)
session.commit()

# Query with relationships
employees = session.query(Employee).filter(
    Employee.department.has(name='Engineering')
).all()

for emp in employees:
    print(f"{emp.name} works in {emp.department.name}")

session.close()
```

--------------------------------

### Configure DECFLOAT for high-precision financial data

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Demonstrates how to use the DECFLOAT type for financial calculations requiring up to 38 digits of precision. It highlights the requirement to enable the parameter via the connection URL.

```python
from sqlalchemy import Column, Integer, MetaData, Table, create_engine
from snowflake.sqlalchemy import DECFLOAT, URL

metadata = MetaData()

financials = Table('financial_data', metadata,
    Column('id', Integer, primary_key=True),
    Column('precise_value', DECFLOAT()),
)

# Enable full 38-digit precision via URL parameter
engine = create_engine(URL(
    account='abc123',
    user='testuser1',
    password='pass',
    database='db',
    enable_decfloat=True,
))

metadata.create_all(engine)
```

--------------------------------

### Snowflake Timestamp Types with SQLAlchemy

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Demonstrates how to use Snowflake's distinct timestamp types (TIMESTAMP_NTZ, TIMESTAMP_TZ, TIMESTAMP_LTZ) and SQLAlchemy's DateTime with timezone support. Shows table definition and insertion of timezone-aware data. Requires `snowflake-sqlalchemy` and `sqlalchemy`.

```python
from sqlalchemy import Column, Integer, MetaData, Table, DateTime, create_engine
from snowflake.sqlalchemy import TIMESTAMP_NTZ, TIMESTAMP_TZ, TIMESTAMP_LTZ, URL
from datetime import datetime, timezone

metadata = MetaData()

events = Table('events', metadata,
    Column('id', Integer, primary_key=True),
    Column('created_at', TIMESTAMP_NTZ()),      # No timezone info stored
    Column('scheduled_at', TIMESTAMP_TZ()),     # Stores timezone with value
    Column('logged_at', TIMESTAMP_LTZ()),       # Converts to session timezone
    Column('updated_at', DateTime(timezone=True)),  # Also produces TIMESTAMP_TZ
)

# Assume 'engine' is already created as shown in the connection example
# engine = create_engine(URL(account='abc123', user='user', password='pass', database='db'))
# metadata.create_all(engine)

# Insert timezone-aware data
# with engine.connect() as conn:
#     conn.execute(events.insert().values(
#         id=1,
#         created_at=datetime(2024, 1, 15, 10, 30, 0),
#         scheduled_at=datetime(2024, 1, 15, 10, 30, 0, tzinfo=timezone.utc),
#         logged_at=datetime.now(),
#     ))
#     conn.commit()
```

--------------------------------

### Alembic Snowflake Dialect Implementation

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Custom Alembic implementation for the Snowflake dialect. This class needs to be added to alembic/env.py before migration functions. It enables Alembic to correctly interact with Snowflake's specific DDL.

```python
from alembic.ddl.impl import DefaultImpl

class SnowflakeImpl(DefaultImpl):
    """Alembic implementation for Snowflake dialect."""
    __dialect__ = 'snowflake'
```

--------------------------------

### Define DECFLOAT Data Type with Snowflake SQLAlchemy

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Demonstrates how to define a table with the `DECFLOAT` data type using Snowflake SQLAlchemy. The `DECFLOAT` type supports decimal floating-point numbers with up to 38 significant digits. This code requires SQLAlchemy and Snowflake SQLAlchemy.

```python
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)
```

--------------------------------

### Define Iceberg Table with MAP Type in Python

Source: https://github.com/snowflakedb/snowflake-sqlalchemy/blob/main/README.md

Illustrates how to define an Iceberg table with a MAP structured data type using Snowflake SQLAlchemy. The MAP type supports specifying key and value types, along with nullability.

```python
from snowflake.sqlalchemy import MAP, OBJECT, ARRAY
from sqlalchemy import Table, Column, Integer, String, MetaData, text, NUMBER, TEXT
from sqlalchemy.schema import CreateTable

# Assuming 'metadata', 'engine', 'table_name', 'external_volume', 'base_location' are defined
# Example for MAP
map_table = Table(
    'iceberg_map_table',
    metadata,
    Column("id", Integer, primary_key=True),
    Column("map_col", MAP(NUMBER(10, 0), TEXT(16777216))),
    # Add other columns as needed
    # For actual Iceberg table creation, you'd use a specific IcebergTable construct if available
)
# metadata.create_all(engine) # This would create a standard SQLAlchemy table, not an Iceberg table directly without specific Iceberg integration
```

--------------------------------

### Implement Hybrid Tables for Transactional Workloads

Source: https://context7.com/snowflakedb/snowflake-sqlalchemy/llms.txt

Utilizes HybridTable to support OLTP operations, including primary key enforcement and index creation. These tables are designed for transactional consistency and fast lookups.

```python
from sqlalchemy import Column, Integer, String, MetaData, Index
from snowflake.sqlalchemy import HybridTable

metadata = MetaData()

# Hybrid table with primary key (required) and indexes
hybrid_users = HybridTable(
    'hybrid_users',
    metadata,
    Column('id', Integer, primary_key=True),  # Primary key required
    Column('email', String(200)),
    Column('name', String(100)),
    Column('department', String(50)),
    # Indexes are only supported on Hybrid Tables
    Index('idx_email', 'email'),
    Index('idx_dept_name', 'department', 'name'),
)

metadata.create_all(engine)

# Insert and query with transactional semantics
with engine.connect() as conn:
    conn.execute(hybrid_users.insert().values(
        id=1, email='alice@example.com', name='Alice', department='Engineering'
    ))
    conn.commit()

    # Fast lookups using indexes
    result = conn.execute(
        hybrid_users.select().where(hybrid_users.c.email == 'alice@example.com')
    )
    print(result.fetchone())
```