Streamlit连接Snowflake：为可编辑表格指定列添加下拉选择

核心思路 #

先从Snowflake拉取目标表数据，提取最后两列的唯一值作为下拉选项，再通过Streamlit表格组件配置这两列为下拉选择框，最后处理编辑后的回写逻辑。

方案一：使用Streamlit原生st.data_editor（无需额外依赖） #

import streamlit as st
import snowflake.connector
import pandas as pd

# Snowflake连接（建议用st.secrets存储敏感配置）
conn = snowflake.connector.connect(
    user=st.secrets["snowflake"]["user"],
    password=st.secrets["snowflake"]["password"],
    account=st.secrets["snowflake"]["account"],
    warehouse=st.secrets["snowflake"]["warehouse"],
    database=st.secrets["snowflake"]["database"],
    schema=st.secrets["snowflake"]["schema"]
)

# 拉取主表数据
table_name = "YOUR_TARGET_TABLE"
df = pd.read_sql(f"SELECT * FROM {table_name}", conn)

# 获取最后两列的唯一下拉选项
last_two_cols = df.columns[-2:]
dropdown_options = {col: sorted(df[col].dropna().unique().tolist()) for col in last_two_cols}

# 配置下拉列
column_config = {}
for col in last_two_cols:
    column_config[col] = st.column_config.SelectboxColumn(
        label=col,
        options=dropdown_options[col],
        required=True,
        width="medium"
    )

# 展示可编辑表格
edited_df = st.data_editor(
    df,
    column_config=column_config,
    num_rows="dynamic",
    use_container_width=True
)

# 回写Snowflake逻辑
if st.button("保存修改"):
    primary_key = df.columns[0]  # 替换为你的主键列名
    cursor = conn.cursor()
    try:
        for _, row in edited_df.iterrows():
            # 对比原始数据，仅更新有变化的行
            original_row = df[df[primary_key] == row[primary_key]]
            if not original_row[last_two_cols].equals(row[last_two_cols]):
                update_sql = f"""
                    UPDATE {table_name}
                    SET {last_two_cols[0]} = %s, {last_two_cols[1]} = %s
                    WHERE {primary_key} = %s
                """
                cursor.execute(update_sql, (row[last_two_cols[0]], row[last_two_cols[1]], row[primary_key]))
        conn.commit()
        st.success("修改已同步到Snowflake！")
    except Exception as e:
        st.error(f"保存失败：{str(e)}")
        conn.rollback()
    finally:
        cursor.close()
        conn.close()

方案二：使用streamlit-aggrid（更灵活的样式控制） #

如果需要更贴近示例截图的UI样式，可使用streamlit-aggrid组件：

1. 先安装依赖：pip install streamlit-aggrid
2. 代码实现：

import streamlit as st
import snowflake.connector
import pandas as pd
from st_aggrid import AgGrid, GridOptionsBuilder

# Snowflake连接（同上）
conn = snowflake.connector.connect(
    user=st.secrets["snowflake"]["user"],
    password=st.secrets["snowflake"]["password"],
    account=st.secrets["snowflake"]["account"],
    warehouse=st.secrets["snowflake"]["warehouse"],
    database=st.secrets["snowflake"]["database"],
    schema=st.secrets["snowflake"]["schema"]
)

# 拉取数据和获取下拉选项（同上）
table_name = "YOUR_TARGET_TABLE"
df = pd.read_sql(f"SELECT * FROM {table_name}", conn)
last_two_cols = df.columns[-2:]
dropdown_options = {col: sorted(df[col].dropna().unique().tolist()) for col in last_two_cols}

# 配置AgGrid下拉列
gb = GridOptionsBuilder.from_dataframe(df)
for col in last_two_cols:
    gb.configure_column(
        col,
        editable=True,
        cellEditor="agSelectCellEditor",
        cellEditorParams={"values": dropdown_options[col]},
        cellStyle={"textAlign": "left"}
    )
grid_options = gb.build()

# 展示表格
grid_response = AgGrid(
    df,
    gridOptions=grid_options,
    editable=True,
    fit_columns_on_grid_load=True,
    use_container_width=True
)
edited_df = grid_response["data"]

# 回写逻辑同方案一
if st.button("保存修改"):
    primary_key = df.columns[0]
    cursor = conn.cursor()
    try:
        for _, row in edited_df.iterrows():
            original_row = df[df[primary_key] == row[primary_key]]
            if not original_row[last_two_cols].equals(row[last_two_cols]):
                update_sql = f"""
                    UPDATE {table_name}
                    SET {last_two_cols[0]} = %s, {last_two_cols[1]} = %s
                    WHERE {primary_key} = %s
                """
                cursor.execute(update_sql, (row[last_two_cols[0]], row[last_two_cols[1]], row[primary_key]))
        conn.commit()
        st.success("修改已同步到Snowflake！")
    except Exception as e:
        st.error(f"保存失败：{str(e)}")
        conn.rollback()
    finally:
        cursor.close()
        conn.close()

注意事项 #

• 敏感配置（账号、密码等）务必用st.secrets存储，不要硬编码
• 如果下拉选项需要实时同步Snowflake数据，每次页面刷新会重新拉取唯一值
• 回写逻辑需根据你的表结构调整主键匹配规则，避免批量更新出错
• 数据量较大时建议添加分页或过滤逻辑，提升页面性能

内容的提问来源于stack exchange，提问作者bobby1985