Snowflake: Python (snowflake-connector-python)

Creates the table, uploads the CSV to a stage with PUT, loads it with COPY INTO and aggregates. Every statement is ordinary Snowflake SQL.

The only change

# host / account URL
 myorg-myaccount.snowflakecomputing.com
 warehouse.internal (MachWarehouse Snowflake REST endpoint, HTTPS 443)
# account
 myorg-myaccount
 the MachWarehouse account name (any label; kept for drivers that require it)
# user / password, role, warehouse, database, schema
 Snowflake values
 MachWarehouse values (warehouse = compute pool)

In this program that is SNOWFLAKE_HOST=warehouse.internal; with Snowflake it is unset (the default <account>.snowflakecomputing.com is used). In connections.toml or SnowSQL config, set host = "warehouse.internal" in the same way.

Run

python3 -m venv .venv && . .venv/bin/activate
pip install -r requirements.txt
SNOWFLAKE_ACCOUNT=demo SNOWFLAKE_HOST=warehouse.internal SNOWFLAKE_USER=app SNOWFLAKE_PASSWORD=... python main.py

Library: snowflake-connector-python 5.1.0.

Source

The program as it is kept and tested in the Mach Database Warehouse repository; highlighted lines read the connection setting.

main.py

"""Load the demo events and print revenue per country with snowflake-connector-python.

Plain Snowflake connector code. The host comes from configuration, as it does
for Snowflake private links and proxies; nothing here knows about MachWarehouse.
"""

import os
import pathlib

import snowflake.connector

DATASET = pathlib.Path(__file__).resolve().parents[2] / "dataset"


def main() -> None:
    account = os.environ["SNOWFLAKE_ACCOUNT"]
    conn = snowflake.connector.connect(
        account=account,
        host=os.environ.get("SNOWFLAKE_HOST", f"{account}.snowflakecomputing.com"),
        port=443,
        protocol="https",
        user=os.environ["SNOWFLAKE_USER"],
        password=os.environ["SNOWFLAKE_PASSWORD"],
        role=os.environ.get("SNOWFLAKE_ROLE", "ANALYST"),
        warehouse=os.environ.get("SNOWFLAKE_WAREHOUSE", "BI"),
        database=os.environ.get("SNOWFLAKE_DATABASE", "DEMO"),
        schema=os.environ.get("SNOWFLAKE_SCHEMA", "PUBLIC"),
    )
    try:
        cur = conn.cursor()
        cur.execute((DATASET / "snowflake.sql").read_text())
        cur.execute("CREATE TEMPORARY STAGE IF NOT EXISTS demo_stage")
        cur.execute(f"PUT file://{DATASET / 'events.csv'} @demo_stage AUTO_COMPRESS=TRUE")
        cur.execute(
            "COPY INTO events FROM @demo_stage "
            "FILE_FORMAT = (TYPE = CSV SKIP_HEADER = 1) ON_ERROR = ABORT_STATEMENT"
        )
        cur.execute(
            "SELECT country, COUNT(*) AS events, SUM(revenue) AS revenue "
            "FROM events GROUP BY country ORDER BY revenue DESC"
        )
        for country, events, revenue in cur:
            print(f"{country}\t{events}\t{revenue}")
        # Arrow result chunks into pandas, as with Snowflake (needs the [pandas] extra):
        # df = cur.execute("SELECT * FROM events").fetch_pandas_all()
    finally:
        conn.close()


if __name__ == "__main__":
    main()

requirements.txt

snowflake-connector-python==5.1.0