Skip to main content

Snowflake Plugin for Flyte

Run Snowflake SQL queries as Flyte tasks with parameterized inputs, key-pair authentication, batch inserts, and DataFrame support.

Installation

pip install flyteplugins-snowflake

Quick start

from flyteplugins.snowflake import Snowflake, SnowflakeConfig

import flyte

config = SnowflakeConfig(
    account="myorg-myaccount",
    user="flyte_user",
    database="ANALYTICS",
    schema="PUBLIC",
    warehouse="COMPUTE_WH",
)

query = Snowflake(
    name="count_users",
    query_template="SELECT COUNT(*) FROM users",
    plugin_config=config,
    snowflake_private_key="snowflake-pk",
)

Authentication

The plugin supports Snowflake key-pair authentication. Pass secret keys via snowflake_private_key (and optionally snowflake_private_key_passphrase).

task = Snowflake(
    name="my_task",
    query_template="SELECT 1",
    plugin_config=config,
    snowflake_private_key="private-key",
    snowflake_private_key_passphrase="passphrase",
    # Generates env vars: PRIVATE_KEY, PASSPHRASE
)

For other auth methods (password, OAuth, etc.), pass them via connection_kwargs:

config = SnowflakeConfig(
    account="myorg-myaccount",
    user="flyte_user",
    database="ANALYTICS",
    schema="PUBLIC",
    warehouse="COMPUTE_WH",
    connection_kwargs={"password": "...", "role": "ADMIN"},
)

Parameterized queries

Use %(name)s placeholders and typed inputs:

lookup = Snowflake(
    name="lookup_user",
    query_template="SELECT * FROM users WHERE id = %(user_id)s",
    plugin_config=config,
    inputs={"user_id": int},
    output_dataframe_type=pd.DataFrame,
    snowflake_private_key="snowflake-pk",
)

Batch inserts

Set batch=True to expand list inputs into multi-row VALUES clauses:

insert_rows = Snowflake(
    name="insert_users",
    query_template="INSERT INTO users (id, name, age) VALUES (%(id)s, %(name)s, %(age)s)",
    plugin_config=config,
    inputs={"id": list[int], "name": list[str], "age": list[int]},
    snowflake_private_key="snowflake-pk",
    batch=True,
)

# Calling with id=[1,2], name=["Alice","Bob"], age=[30,25] expands to:
# INSERT INTO users (id, name, age) VALUES (%(id_0)s, %(name_0)s, %(age_0)s), (%(id_1)s, %(name_1)s, %(age_1)s)

Reading results as DataFrames

Set output_dataframe_type to get query results as a pandas DataFrame:

import pandas as pd

select_task = Snowflake(
    name="get_users",
    query_template="SELECT * FROM users",
    plugin_config=config,
    output_dataframe_type=pd.DataFrame,
    snowflake_private_key="snowflake-pk",
)

Full example

import pandas as pd
from flyteplugins.snowflake import Snowflake, SnowflakeConfig

import flyte

config = SnowflakeConfig(
    user="KEVIN",
    account="PWGJLTH-XKB21544",
    database="FLYTE",
    schema="PUBLIC",
    warehouse="COMPUTE_WH",
)

insert_task = Snowflake(
    name="insert_rows",
    inputs={"id": list[int], "name": list[str], "age": list[int]},
    plugin_config=config,
    query_template="INSERT INTO FLYTE.PUBLIC.TEST (ID, NAME, AGE) VALUES (%(id)s, %(name)s, %(age)s)",
    snowflake_private_key="snowflake",
    batch=True,
)

select_task = Snowflake(
    name="select_all",
    output_dataframe_type=pd.DataFrame,
    plugin_config=config,
    query_template="SELECT * FROM FLYTE.PUBLIC.TEST",
    snowflake_private_key="snowflake",
)

snowflake_env = flyte.TaskEnvironment.from_task("snowflake_env", insert_task, select_task)

env = flyte.TaskEnvironment(
    name="example_env",
    image=flyte.Image.from_debian_base().with_pip_packages("flyteplugins-snowflake"),
    secrets=[flyte.Secret(key="snowflake", as_env_var="SNOWFLAKE_PRIVATE_KEY")],
    depends_on=[snowflake_env],
)


@env.task
async def main(ids: list[int], names: list[str], ages: list[int]) -> float:
    await insert_task(id=ids, name=names, age=ages)
    df = await select_task()
    return df["AGE"].mean().item()


if __name__ == "__main__":
    flyte.init_from_config()
    run = flyte.with_runcontext(mode="remote").run(
        main, ids=[123, 456], names=["Kevin", "Alice"], ages=[30, 25],
    )
    print(run.url)

Release files for flyteplugins-snowflake 2.8.0

For a detailed explanation of source distributions (sdists) and built distributions (wheels), please see the package formats documentation.

Built distribution (wheel)

Table of built distributions (wheels) for flyteplugins-snowflake 2.8.0
File Interpreter ABI Platform
flyteplugins_snowflake-2.8.0-py3-none-any.whl Python 3 none any Details

Release files / flyteplugins_snowflake-2.8.0-py3-none-any.whl

Download URL flyteplugins_snowflake-2.8.0-py3-none-any.whl
Size 11.3 kB
Tags Python 3
SHA-256 checksum
How to use checksums
0f756e08ea85c4bdd058ef66798c054ff7bba31531599e92b63dad0bd3ae6c19
BLAKE2b-256 checksum
How to use checksums
50e2a5203f309b6e4631ce9965c3832c312ee7989aa9f8d9254836446283b46b
Upload date
Uploaded using Trusted Publishing?
What is trusted publishing?
No
Uploaded via twine/7.0.0 CPython/3.13.15

Release history Release notifications | RSS feed

2.10.1

1 release file

2.10.0

1 release file

2.9.0

1 release file

2.8.1

1 release file

This release

2.8.0 This release

1 release file

2.7.2

1 release file

2.7.1

1 release file

2.7.0

1 release file

2.6.13

1 release file

2.6.12

1 release file

2.6.11

1 release file

2.6.10

1 release file

2.6.9

1 release file

2.6.8

1 release file

2.6.7

1 release file

2.6.6

1 release file

2.6.5

1 release file

2.6.4

1 release file

2.6.3

1 release file

2.6.2

1 release file

2.6.1

1 release file

2.6.0

1 release file

2.5.20

1 release file

2.5.18

1 release file

2.5.17

1 release file

2.5.14

1 release file

2.5.13

1 release file

2.5.12

1 release file

2.5.11

1 release file

2.5.10

1 release file

2.5.9

1 release file

2.5.8

1 release file

2.5.7

1 release file

2.5.6

1 release file

2.5.5

1 release file

2.5.4

1 release file

2.5.3

1 release file

2.5.2

1 release file

2.5.1

1 release file

2.5.0

1 release file

2.4.4

1 release file

2.4.3

1 release file

2.4.2

1 release file

2.4.1

1 release file

2.4.0

1 release file

2.3.9

1 release file

2.3.8

1 release file

2.3.7

1 release file

2.3.6

1 release file

2.3.5

1 release file

2.3.4

1 release file

2.3.3

1 release file

2.3.2

1 release file

2.3.1

1 release file

2.3.0

1 release file

2.2.4

1 release file

2.2.3

1 release file

2.2.2

1 release file

2.2.1

1 release file

2.2.0

1 release file

2.1.9

1 release file

2.1.8

1 release file

2.1.7

1 release file

2.1.6

1 release file

2.1.5

1 release file

2.1.4

1 release file

2.1.3

1 release file

2.1.2

1 release file

2.1.1

1 release file

2.1.0

1 release file

2.0.12

1 release file

2.0.11

1 release file

2.0.10

1 release file

2.0.9

1 release file

2.0.8

1 release file

2.0.7

1 release file

2.0.6

1 release file

2.0.4

1 release file

2.0.3

1 release file

2.0.2

1 release file

2.0.1

1 release file

2.0.0

1 release file

Anthropic, PBC Visionary sponsor Bloomberg Visionary sponsor Hudson River Trading Visionary sponsor Meta Visionary sponsor NVIDIA Visionary sponsor Microsoft Sustainability sponsor Depot Continuous Integration AWS Cloud computing and Security Sponsor Datadog Monitoring Fastly CDN Google Download Analytics Sentry Error logging StatusPage Status page