connecting-streamlit-to-snowflake

用途

本技能面向需要将 Streamlit 应用与 Snowflake 数据库对接的场景，涵盖建立数据库连接、管理凭据密钥，以及在 Streamlit 应用中查询 Snowflake 数据等任务。

主要能力

使用 st.connection

文档强调应始终使用 st.connection("snowflake")，而不是原始连接器。示例：

import streamlit as st

conn = st.connection("snowflake")

# Query data
df = conn.query("SELECT * FROM my_table LIMIT 100")
st.dataframe(df)

使用 st.connection 的理由包括：自动连接池、内置缓存、处理重连、可与 st.secrets 配合。

调用者权限连接

对于在 Snowflake 中运行的应用，可使用调用者权限（Streamlit 1.53+），以查看者的权限而非应用所有者的权限执行查询：

conn = st.connection("snowflake", type="snowflake-callers-rights")

适用场景包括：不同用户应基于其 Snowflake 角色看到不同数据；希望行级安全基于查看者生效；不希望应用拥有提升的权限。

缓存查询

使用内置 ttl 参数缓存查询结果：

from datetime import timedelta

conn = st.connection("snowflake")

# Cache for 10 minutes
df = conn.query("SELECT * FROM metrics", ttl=timedelta(minutes=10))

# Cache for 1 hour
df = conn.query("SELECT * FROM reference_data", ttl=3600)

使用 st.secrets 配置

将凭据存放在 .streamlit/secrets.toml 中（切勿提交该文件）。文档强调：account 和 host 的值必须来自用户的 Snowflake CLI 连接配置，运行 snow connection list 并使用其确切值；account 错误会重定向到错误的登录页面。

# .streamlit/secrets.toml
[connections.snowflake]
account = "ORGNAME-ACCTNAME"            # from `snow connection list`
host = "myaccount.snowflakecomputing.com"  # from `snow connection list` (include if present)
user = "your_user"
authenticator = "externalbrowser"
warehouse = "your_warehouse"
database = "your_database"
schema = "your_schema"

并加入 .gitignore：

.streamlit/secrets.toml

参数化查询

使用参数防止 SQL 注入：

conn = st.connection("snowflake")

# Safe: parameterized
df = conn.query(
    "SELECT * FROM users WHERE region = :region",
    params={"region": selected_region}
)

# UNSAFE: string formatting - don't do this
# df = conn.query(f"SELECT * FROM users WHERE region = '{selected_region}'")

写入数据

使用 session 执行写操作：

conn = st.connection("snowflake")
session = conn.session()

# Write a dataframe
session.write_pandas(df, "MY_TABLE", auto_create_table=True)

# Execute statements
session.sql("INSERT INTO logs VALUES (:ts, :msg)", params={...}).collect()

多个连接

可在 secrets 中定义多个连接：

# .streamlit/secrets.toml
[connections.snowflake]
account = "prod_account"
# ... prod credentials

[connections.snowflake_staging]
account = "staging_account"
# ... staging credentials

prod_conn = st.connection("snowflake")
staging_conn = st.connection("snowflake_staging")

与 Cortex 聊天

可基于 Snowflake Cortex LLM 构建聊天界面：

import streamlit as st
from snowflake.cortex import complete

st.set_page_config(page_title="AI Assistant", page_icon=":sparkles:")

if "messages" not in st.session_state:
    st.session_state.messages = []

for msg in st.session_state.messages:
    with st.chat_message(msg["role"]):
        st.write(msg["content"])

if prompt := st.chat_input("Ask anything"):
    st.session_state.messages.append({"role": "user", "content": prompt})

    with st.chat_message("user"):
        st.write(prompt)

    with st.chat_message("assistant"):
        response = st.write_stream(
            complete(
                "claude-3-5-sonnet",
                prompt,
                session=st.connection("snowflake").session(),
                stream=True,
            )
        )

    st.session_state.messages.append({"role": "assistant", "content": response})

更多聊天模式（头像、建议、历史管理）可参见 building-streamlit-chat-ui。

使用方式与限制

Python 3.12+ 依赖注意事项

streamlit[snowflake] 将 snowflake-connector-python 限定在 python_version < "3.12"。在 Python 3.12+ 上，该连接器会被静默跳过，运行时会出现 No module named 'snowflake'。应始终在 pyproject.toml 中显式添加 snowflake-connector-python>=3.3.0 作为依赖：

dependencies = [
    "snowflake-connector-python>=3.3.0",
    "streamlit[snowflake]>=1.54.0",
]

参考

st.connection、SnowflakeConnection、st.secrets。