Example 1: Configuration with Streamlit secrets

You can configure your Snowflake connection using Streamlit's Secrets management. For example, if you have MFA enabled on your account, you can connect using key-pair authentication.

[connections.snowflake]
account = "xxx-xxx"
user = "xxx"
private_key_file = "/xxx/xxx/xxx.p8"
role = "xxx"
warehouse = "xxx"
database = "xxx"
schema = "xxx"

import streamlit as st

conn = st.connection("snowflake")
df = conn.query("SELECT * FROM my_table")

Example 2: Configuration with keyword arguments and external authentication

You can configure your Snowflake connection with keyword arguments. The keyword arguments are merged with (and take precedence over) the values in secrets.toml. However, if you name your connection "snowflake" and don't have a [connections.snowflake] dictionary in your secrets.toml file, Streamlit will ignore any keyword arguments and use the default Snowflake connection as described in Example 5 and Example 6. To configure your connection using only keyword arguments, declare a name for the connection other than "snowflake".

For example, if your Snowflake account supports SSO, you can set up a quick local connection for development using browser-based SSO. Because there is nothing configured in secrets.toml, the name is an empty string and the type is set to "snowflake". This prevents Streamlit from ignoring the keyword arguments and using a default Snowflake connection.

import streamlit as st

conn = st.connection(
    "",
    type="snowflake",
    account="xxx-xxx",
    user="xxx",
    authenticator="externalbrowser",
)
df = conn.query("SELECT * FROM my_table")

Example 3: Named connection with Snowflake's connection configuration file

Snowflake's Python Connector supports a connection configuration file, which is well integrated with Streamlit's SnowflakeConnection. If you already have one or more connections configured, all you need to do is pass the name of the connection to use.

[my_connection]
account = "xxx-xxx"
user = "xxx"
password = "xxx"
warehouse = "xxx"
database = "xxx"
schema = "xxx"

import streamlit as st

conn = st.connection("my_connection", type="snowflake")
df = conn.query("SELECT * FROM my_table")

Example 4: Named connection with Streamlit secrets and Snowflake's connection configuration file

If you have a Snowflake configuration file with a connection named my_connection as in Example 3, you can pass the connection name through secrets.toml.

[connections.snowflake]
connection_name = "my_connection"

import streamlit as st

conn = st.connection("snowflake")
df = conn.query("SELECT * FROM my_table")

Example 5: Default connection with an environment variable

If you don't have a [connections.snowflake] dictionary in your secrets.toml file and use st.connection("snowflake"), Streamlit will use the default connection for the Snowflake Python Connector.

If you have a Snowflake configuration file with a connection named my_connection as in Example 3, you can set an environment variable to declare it as the default Snowflake connection.

SNOWFLAKE_DEFAULT_CONNECTION_NAME = "my_connection"

import streamlit as st

conn = st.connection("snowflake")
df = conn.query("SELECT * FROM my_table")

Example 6: Default connection in Snowflake's connection configuration file

If you have a Snowflake configuration file that defines your default connection, Streamlit will automatically use it if no other connection is declared.

[default]
account = "xxx-xxx"
user = "xxx"
password = "xxx"
warehouse = "xxx"
database = "xxx"
schema = "xxx"

import streamlit as st

conn = st.connection("snowflake")
df = conn.query("SELECT * FROM my_table")

Example 7: Caller's rights connection when running in Snowpark Container Services

You can use "snowflake-callers-rights" type connections in any environment running on Snowpark Container Services, including Streamlit in Snowflake on containers and any self-managed caller's rights Service.

This will use the Snowpark-provided account, host, database, and schema to connect. Additionally, it will set client_session_keep_alive to True. These values may be overridden with **kwargs in st.connection. For a complete list of keyword arguments, see the documentation for snowflake.connector.connect().

import streamlit as st

conn = st.connection("snowflake-callers-rights")
df = conn.query("SELECT * FROM my_table")

If you want to develop locally with a caller's rights connection, use an environment variable to logically switch between a "snowflake" connection locally and a "snowflake-callers-rights" connection in Snowpark Container Services.

import streamlit as st

conn = (
    st.connection("snowflake")
    if "LOCAL_DEVELOPMENT" in st.secrets and st.secrets["LOCAL_DEVELOPMENT"]
    else st.connection("snowflake-callers-rights")
)
df = conn.query("SELECT * FROM my_table")

The following example uses a cursor to insert multiple rows into a table. The qmark parameter style is specified as an optional keyword argument. Alternatively, the parameter style can be declared in your connection configuration file. For more information, see the Snowflake Connector for Python documentation.

import streamlit as st

conn = st.connection("snowflake", "paramstyle"="qmark")
rows_to_insert = [("Mary", "dog"), ("John", "cat"), ("Robert", "bird")]
conn.cursor().executemany(
    "INSERT INTO mytable (name, pet) VALUES (?, ?)", rows_to_insert
)

The following example uses a cursor to submit an asynchronous query, saves the query ID, then periodically checks the query status through the connection before retrieving the results.

import streamlit as st
import time

conn = st.connection("snowflake")
cur = conn.cursor()
cur.execute_async("SELECT * FROM my_table")
query_id = cur.sfqid
while True:
    status = conn.raw_connection.get_query_status(query_id)
    if conn.raw_connection.is_still_running(status):
        time.sleep(1)
    else:
        break
cur.get_results_from_sfqid(query_id)
df = cur.fetchall()

The following example creates a new Snowpark session and uses it to run a query.

import streamlit as st

conn = st.connection("snowflake")
session = conn.session()
df = session.sql("SELECT * FROM my_table").collect()

The following example uses the database and schema currently in use in the session and copies the data into a table named "my_table."

import streamlit as st
import pandas as pd

df = pd.DataFrame(
    {"Name": ["Mary", "John", "Robert"], "Pet": ["dog", "cat", "bird"]}
)
conn = st.connection("snowflake")
conn.write_pandas(df, "my_table")