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.

.streamlit/secrets.toml:

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

Your app code:

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 (with or without secrets.toml). For example, if your Snowflake account supports SSO, you can set up a quick local connection for development using browser-based SSO.

import streamlit as st
conn = st.connection(
    "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.

~/.snowflake/connections.toml:

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

Your app code:

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.

.streamlit/secrets.toml:

[connections.snowflake]
connection_name = "my_connection"

Your app code:

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 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"

Your app code:

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.

~/.snowflake/connections.toml:

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

Your app code:

import streamlit as st
conn = st.connection("snowflake")
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
)

import streamlit as st

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

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()

import streamlit as st

conn = st.connection("my_conn")

# Reset the connection before using it if it isn't healthy
# Note: is_healthy() isn't a real method and is just shown for example here.
if not conn.is_healthy():
    conn.reset()

# Do stuff with conn...

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")