Snowflake Streamlit应用数据更新功能失效问题求助

云原生阿岚

2026-6-22

Snowflake Streamlit应用数据更新功能失效问题排查

问题说明

开发的Snowflake Streamlit应用支持数据表回写，插入和搜索功能正常，但数据更新功能未达预期：

初始实现中，点击更新按钮后数据表无变化
调整将update_button判断移入表单内部后，程序始终未进入if update_button:分支，更新操作完全不执行

初始代码实现

# Section for searching and updating existing records
st.subheader("Search for Record to Update")
with st.form("search_record"):
    search_by = st.radio("Search by:", ('MCC Code', 'MCC_TITLE'))
    search_value = st.text_input(f"Enter {search_by}:")
    search_button = st.form_submit_button('Search')

# Display and update form if record is found
if search_button:
    try:
        search_column = '"MCC Code:"' if search_by == 'MCC Code' else '"MCC_TITLE"'
        query = f"""
        SELECT * FROM IK_DEV.POWER_WRITE_BACK.MERCHANT_CATEGORY_CODE_LIST
        WHERE {search_column} = '{search_value}'
        """
        df_search_result = session.sql(query).to_pandas()
        if not df_search_result.empty:
            st.success("Record found!", icon="✅")
            record = df_search_result.iloc[0]

            with st.form("update_record"):
                mcc_code = st.text_input('MCC Code:', value=record['MCC Code:'])
                mcc_title = st.text_input('MCC Title:', value=record['MCC_TITLE'])
                ikhokha_defined = st.text_input('iKhokha Defined:', value=record['iKhokha defined (What our customers see)'])
                mcc_description = st.text_area('MCC Description:', value=record['MCC_DESCRIPTION'])
                included_in_this_mcc = st.text_input('Included in this MCC:', value=record['INCLUDED_IN_THIS_MCC'])
                similar_merchants = st.text_input('Similar Merchants:', value=record['SIMILAR_MERCHANTS'])
                restricted_cp = st.checkbox('Restricted CP', value=record['RESTRICTED_CP'] == 1)
                restricted_cnp = st.checkbox('Restricted CNP', value=record['RESTRICTED_CNP'] == 1)
                update_button = st.form_submit_button('Update')

            if update_button:
                try:
                    restricted_cp_int = 1 if restricted_cp else 0
                    restricted_cnp_int = 1 if restricted_cnp else 0

                    update_query = f"""
                    UPDATE IK_DEV.POWER_WRITE_BACK.MERCHANT_CATEGORY_CODE_LIST
                    SET "MCC Code:" = '{mcc_code}', "MCC_TITLE" = '{mcc_title}', "iKhokha defined (What our customers see)" = '{ikhokha_defined}', 
                        "MCC_DESCRIPTION" = '{mcc_description}', "INCLUDED_IN_THIS_MCC" = '{included_in_this_mcc}', "SIMILAR_MERCHANTS" = '{similar_merchants}', 
                        "RESTRICTED_CP" = {restricted_cp_int}, "RESTRICTED_CNP" = {restricted_cnp_int}
                    WHERE {search_column} = '{search_value}'
                    """
                    session.sql(update_query).collect()
                    st.success('Record successfully updated!', icon="✅")
                except Exception as e:
                    st.error(f"Error updating record: {e}")
        else:
            st.error("No record found!", icon="❌")
    except Exception as e:
        st.error(f"Error searching for record: {e}")

调整后代码

# Display and update form if record is found
if search_button:
    try:
        search_column = '"MCC Code:"' if search_by == 'MCC Code' else '"MCC_TITLE"'
        query = f"""
        SELECT * FROM IK_DEV.POWER_WRITE_BACK.MERCHANT_CATEGORY_CODE_LIST
        WHERE {search_column} = '{search_value}'
        """
        df_search_result = session.sql(query).to_pandas()
        if not df_search_result.empty:
            st.success("Record found!", icon="✅")
            record = df_search_result.iloc[0]

            with st.form("update_record"):
                mcc_code = st.text_input('MCC Code:', value=record['MCC Code:'])
                mcc_title = st.text_input('MCC Title:', value=record['MCC_TITLE'])
                ikhokha_defined = st.text_input('iKhokha Defined:', value=record['iKhokha defined (What our customers see)'])
                mcc_description = st.text_area('MCC Description:', value=record['MCC_DESCRIPTION'])
                included_in_this_mcc = st.text_input('Included in this MCC:', value=record['INCLUDED_IN_THIS_MCC'])
                similar_merchants = st.text_input('Similar Merchants:', value=record['SIMILAR_MERCHANTS'])
                restricted_cp = st.checkbox('Restricted CP', value=record['RESTRICTED_CP'] == 1)
                restricted_cnp = st.checkbox('Restricted CNP', value=record['RESTRICTED_CNP'] == 1)
                update_button = st.form_submit_button('Update')

                if update_button:
                    try:
                        restricted_cp_int = 1 if restricted_cp else 0
                        restricted_cnp_int = 1 if restricted_cnp else 0

                        update_query = f"""
                        UPDATE IK_DEV.POWER_WRITE_BACK.MERCHANT_CATEGORY_CODE_LIST
                        SET "MCC Code:" = '{mcc_code}', "MCC_TITLE" = '{mcc_title}', "iKhokha defined (What our customers see)" = '{ikhokha_defined}', 
                        "MCC_DESCRIPTION" = '{mcc_description}', "INCLUDED_IN_THIS_MCC" = '{included_in_this_mcc}', "SIMILAR_MERCHANTS" = '{similar_merchants}', 
                        "RESTRICTED_CP" = {restricted_cp_int}, "RESTRICTED_CNP" = {restricted_cnp_int}
                        WHERE {search_column} = '{search_value}'
                        """

                        st.write(update_query)
                        result_df = session.sql(update_query).collect()
                        st.dataframe(result_df)
                        st.success('Record successfully updated!', icon="✅")
                    except Exception as e:
                        st.error(f"Error updating record: {e}")
        else:
            st.error("No record found!", icon="❌")
    except Exception as e:
        st.error(f"Error searching for record: {e}")

问题根源分析

Streamlit页面刷新机制限制：
Streamlit是单页应用，点击任何表单的提交按钮都会触发整个页面重新运行。当点击update_button时，页面刷新，search_button的状态会重置为False，导致if search_button:分支不再执行，更新表单根本不会被渲染，自然无法进入if update_button:分支。
SQL拼接的潜在问题：
用f-string直接拼接SQL语句存在SQL注入风险，同时如果字段值包含单引号，会直接导致SQL语法错误，更新失败。
更新条件的逻辑漏洞：
更新语句的WHERE条件使用的是搜索时的search_value，如果用户修改了作为搜索条件的字段（比如MCC Code），会导致更新条件不匹配，无法找到目标记录。

修复方案

使用st.session_state保存搜索到的记录，让更新逻辑不依赖search_button的状态，同时改用参数化查询避免SQL注入：

# Section for searching and updating existing records
st.subheader("Search for Record to Update")
with st.form("search_record"):
    search_by = st.radio("Search by:", ('MCC Code', 'MCC_TITLE'))
    search_value = st.text_input(f"Enter {search_by}:")
    search_button = st.form_submit_button('Search')

# Handle search and save record to session state
if search_button:
    try:
        # 使用参数化查询避免SQL注入
        search_column = 'MCC Code:' if search_by == 'MCC Code' else 'MCC_TITLE'
        query = f"""
        SELECT * FROM IK_DEV.POWER_WRITE_BACK.MERCHANT_CATEGORY_CODE_LIST
        WHERE "{search_column}" = %s
        """
        df_search_result = session.sql(query, params=[search_value]).to_pandas()
        if not df_search_result.empty:
            st.success("Record found!", icon="✅")
            # 将查询结果存入session_state，页面刷新后仍能保留
            st.session_state['update_record'] = df_search_result.iloc[0]
        else:
            st.error("No record found!", icon="❌")
            # 清空session_state中的记录
            st.session_state.pop('update_record', None)
    except Exception as e:
        st.error(f"Error searching for record: {e}")
        st.session_state.pop('update_record', None)

# 显示更新表单（依赖session_state中的记录，而非search_button状态）
if 'update_record' in st.session_state:
    record = st.session_state['update_record']
    with st.form("update_record_form"):
        mcc_code = st.text_input('MCC Code:', value=record['MCC Code:'])
        mcc_title = st.text_input('MCC Title:', value=record['MCC_TITLE'])
        ikhokha_defined = st.text_input('iKhokha Defined:', value=record['iKhokha defined (What our customers see)'])
        mcc_description = st.text_area('MCC Description:', value=record['MCC_DESCRIPTION'])
        included_in_this_mcc = st.text_input('Included in this MCC:', value=record['INCLUDED_IN_THIS_MCC'])
        similar_merchants = st.text_input('Similar Merchants:', value=record['SIMILAR_MERCHANTS'])
        restricted_cp = st.checkbox('Restricted CP', value=record['RESTRICTED_CP'] == 1)
        restricted_cnp = st.checkbox('Restricted CNP', value=record['RESTRICTED_CNP'] == 1)
        update_button = st.form_submit_button('Update')

    if update_button:
        try:
            restricted_cp_int = 1 if restricted_cp else 0
            restricted_cnp_int = 1 if restricted_cnp else 0

            # 使用原始记录的主键（假设MCC Code是主键）作为WHERE条件，避免用户修改后匹配失败
            update_query = """
            UPDATE IK_DEV.POWER_WRITE_BACK.MERCHANT_CATEGORY_CODE_LIST
            SET "MCC Code:" = %s, "MCC_TITLE" = %s, "iKhokha defined (What our customers see)" = %s, 
                "MCC_DESCRIPTION" = %s, "INCLUDED_IN_THIS_MCC" = %s, "SIMILAR_MERCHANTS" = %s, 
                "RESTRICTED_CP" = %s, "RESTRICTED_CNP" = %s
            WHERE "MCC Code:" = %s
            """
            # 参数化传递，避免SQL注入
            session.sql(update_query, params=[
                mcc_code, mcc_title, ikhokha_defined,
                mcc_description, included_in_this_mcc, similar_merchants,
                restricted_cp_int, restricted_cnp_int, record['MCC Code:']
            ]).collect()
            
            st.success('Record successfully updated!', icon="✅")
            # 更新session_state中的记录，保持表单数据同步
            st.session_state['update_record'] = {
                'MCC Code:': mcc_code,
                'MCC_TITLE': mcc_title,
                'iKhokha defined (What our customers see)': ikhokha_defined,
                'MCC_DESCRIPTION': mcc_description,
                'INCLUDED_IN_THIS_MCC': included_in_this_mcc,
                'SIMILAR_MERCHANTS': similar_merchants,
                'RESTRICTED_CP': restricted_cp_int,
                'RESTRICTED_CNP': restricted_cnp_int
            }
        except Exception as e:
            st.error(f"Error updating record: {e}")

关键修复点

用st.session_state存储搜索到的记录，确保页面刷新后更新表单仍能渲染
改用参数化SQL查询，彻底避免SQL注入和语法错误
更新条件使用原始记录的主键值，避免用户修改搜索字段后无法匹配目标记录
解耦搜索逻辑和更新逻辑，不再依赖search_button的临时状态

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

Snowflake Streamlit应用数据更新功能失效问题求助

问题说明 #

初始代码实现 #

调整后代码 #

问题根源分析 #

修复方案 #