How to execute a multi-statement Snowflake SQL script (DECLARE, DDL, DML, CASE) from Python with parameters and fetch SELECT results?

I’m new to Python and Snowflake’s scripting blocks. I want to keep a multi-statement Snowflake SQL script in a .sql file and execute it from Python with parameters. The script may include DECLARE, DDL, DML, CASE logic — basically SP-like behavior — and I want to capture any SELECT result sets if present.

Example script (simplified):

BEGIN
    -- Declare a variable using Snowflake scripting
    DECLARE v_admin STRING DEFAULT :admin_user;

    -- Create a temporary table
    CREATE TEMP TABLE temp_data (
        tenant STRING,
        client STRING,
        user_id STRING,
        use_mm STRING,
        access_level STRING
    );

    -- Two inserts
    INSERT INTO temp_data (tenant, client, user_id, use_mm, access_level)
    VALUES
        ('101', '202', 'admin_user', 'true',  NULL),
        ('102', '203', 'guest_user', 'false', NULL);

    -- CASE update using the declared variable
    UPDATE temp_data
    SET access_level = CASE
        WHEN use_mm = 'true' AND user_id = v_admin THEN 'full'
        WHEN use_mm = 'true' THEN 'limited'
        ELSE 'none'
    END;

    -- Return results
    SELECT * FROM temp_data;
END;

What I need:

  • Execute the file from Python, bind parameters (e.g., admin_user), and support multiple statements.
  • If the script contains one or more SELECTs, fetch results (ideally the last SELECT); otherwise just execute.
  • Keep temp tables ephemeral (session-scoped).

Questions:

  1. Can I achieve stored-procedure-like behavior by keeping the SQL in a file and running it from Python?
  2. What’s the recommended way to pass parameters into a scripting block (e.g., :name vs %(name)s vs session variables like SET and $VAR)?
  3. How can I iterate over multiple result sets to capture SELECT outputs?
  4. Any gotchas when mixing DECLARE with DDL/DML inside BEGIN…END (e.g., syntax errors with CREATE TABLE)?

What I’ve tried:

  • Using the Snowflake Python connector’s cursor.execute(sql, params, num_statements=0) with a multi-statement script.
  • A scripting block with DECLARE followed by DDL/DML and a final SELECT.

Environment:

  • Snowflake
  • Python 3.14 with snowflake-connector-python

Thanks for any guidance or example code!

from snowflake import connector

# --- Connection Setup ---
conn = connector.connect(
    user='YOUR_USERNAME',
    password='YOUR_PASSWORD',
    account='YOUR_ACCOUNT_ID',
    warehouse='YOUR_WAREHOUSE',
    database='YOUR_DATABASE',
    schema='YOUR_SCHEMA',
    autocommit=False  # Important for rollback to work
)
cur = conn.cursor()

# --- Multiple Parameters ---
parameters = {
    "tenant": "AL96508",
    "admin_user": "rudresh"
}

try:
    # --- Set all parameters dynamically ---
    for key, value in parameters.items():
        cur.execute(f"SET {key} = '{value}';")

    # --- Multi-statement Snowflake Scripting Block ---
    sql_block = """
    BEGIN
        DECLARE v_admin STRING DEFAULT $admin_user;
        DECLARE v_tenant STRING DEFAULT $tenant;

        CREATE TEMP TABLE temp_data (
            tenant STRING,
            client STRING,
            user_id STRING,
            use_mm STRING,
            access_level STRING
        );

        INSERT INTO temp_data (tenant, client, user_id, use_mm, access_level)
        VALUES
            (v_tenant, '202', v_admin, 'true',  NULL),
            ('102', '203', 'guest_user', 'false', NULL);

        -- Example: Uncomment to test rollback
        -- SELECT 1 / 0;

        UPDATE temp_data
        SET access_level = CASE
            WHEN use_mm = 'true' AND user_id = v_admin THEN 'full'
            WHEN use_mm = 'true' THEN 'limited'
            ELSE 'none'
        END;

        SELECT * FROM temp_data;
    END;
    """

    # --- Execute Script ---
    cur.execute(sql_block)

    # --- Fetch headers and rows ---
    headers = [desc[0] for desc in cur.description]
    rows = cur.fetchall()

    # --- Commit if all good ---
    conn.commit()

    # --- Print results ---
    print(" | ".join(headers))
    print("-" * (len(headers) * 15))
    for row in rows:
        print(" | ".join(str(x) if x is not None else "" for x in row))

except Exception as e:
    print(f"Error occurred: {e}")
    print("Rolling back transaction...")
    conn.rollback()

finally:
    cur.close()
    conn.close()
Вернуться на верх