Как выполнить SQL-скрипт Snowflake с несколькими операторами (DECLARE, DDL, DML, CASE) из Python с параметрами и получить результаты SELECT?

Я новичок в Python и скриптовых блоках Snowflake. Я хочу сохранить SQL-скрипт Snowflake с несколькими инструкциями в файле .sql и запустить его из Python с параметрами. Сценарий может включать в себя DECLARE, DDL, DML, логику CASE — в основном поведение, подобное SP, - и я хочу фиксировать любые выбранные результирующие наборы, если они присутствуют.

Пример сценария (упрощенный):

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;

Что мне нужно:

  • Запустите файл с помощью Python, привяжите параметры (например, admin_user) и поддерживайте несколько инструкций.
  • Если скрипт содержит одну или несколько выборок, извлеките результаты (в идеале последнюю выборку); в противном случае просто выполните.
  • Сохраняйте временные таблицы временными (в рамках сеанса).

Вопросов:

  1. Могу ли я добиться поведения, подобного хранимой процедуре, сохранив SQL в файле и запустив его из Python?
  2. Каков рекомендуемый способ передачи параметров в блок сценариев (например, name против %(name)s против переменных сеанса, таких как SET и $VAR)?
  3. Как я могу выполнить итерацию по нескольким результирующим наборам, чтобы получить выбранные выходные данные?
  4. Есть какие-нибудь ошибки при смешивании DECLARE с DDL/DML внутри BEGIN...END (например, синтаксические ошибки при создании ТАБЛИЦЫ)?

Что я пробовал:

  • Используя курсор Snowflake Python connector.выполните(sql, params, num_statements=0) скрипт с несколькими инструкциями.
  • Блок сценария с ОБЪЯВЛЕНИЕМ, за которым следует DDL/DML и окончательный ВЫБОР.

Окружение:

  • Snowflake
  • Python 3.14 с snowflake-connector-python

Спасибо за любые рекомендации или пример кода!

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()
Вернуться на верх