Snowflake Native Apps中Streamlit无法访问指定数据库的问题

核心问题：会话获取逻辑错误 #

你的代码中使用get_active_session()获取Snowflake会话，这是Snowflake Native Apps环境提供的内置应用会话，该会话使用的是应用部署时指定的运行角色（而非用户输入的凭证）。因此，用户输入的账号密码并未实际用于创建会话，导致应用运行时使用的是应用角色的权限，而非用户自身的权限，这就是为什么你在工作表中能正常查询，但应用中提示权限不足或数据库不存在的原因。

权限授予方案的正确性与永久性 #

如果你给应用的运行角色授予了目标数据库的访问权限，这个授权是永久有效的（除非手动撤销），但这种方案仅适用于让所有使用该应用的用户访问同一固定数据库的场景。如果你的需求是让用户通过自己的凭证访问各自的数据库，那么这个方案并不合适，因为它无法动态切换到用户自身的权限上下文。

代码修复方案 #

需要修改会话获取逻辑，使用用户输入的凭证创建独立会话，而非依赖Native Apps的内置会话。以下是修正后的代码：

import streamlit as st
import pandas as pd
from snowflake.snowpark import Session
from snowflake.snowpark.context import get_active_session

def get_snowflake_session() -> Session:
    try:
        # 使用用户输入的凭证创建会话，而非依赖内置会话
        if all(key in st.session_state for key in ['user', 'password', 'account', 'warehouse', 'database', 'schema']):
            connection_params = {
                "user": st.session_state['user'],
                "password": st.session_state['password'],
                "account": st.session_state['account'],
                "warehouse": st.session_state['warehouse'],
                "database": st.session_state['database'],
                "schema": st.session_state['schema']
            }
            session = Session.builder.configs(connection_params).create()
            return session
        else:
            st.error("缺少必要的连接参数")
            return None
    except Exception as e:
        st.error(f"创建Snowflake会话失败: {e}")
        return None

def show_data_page():
    st.title("Snowflake Data Viewer")

    # 检查是否已存储凭证
    if 'user' in st.session_state:
        try:
            session = get_snowflake_session()
            if session:
                st.success("成功连接到Snowflake")

                # 显示当前数据库和Schema
                st.write(f"数据库: {st.session_state['database']}")
                st.write(f"Schema: {st.session_state['schema']}")

                # 查询指定Schema中的表
                query = f"""
                    SELECT TABLE_NAME FROM INFORMATION_SCHEMA.TABLES 
                    WHERE TABLE_SCHEMA = '{st.session_state['schema'].upper()}'
                    AND TABLE_CATALOG = '{st.session_state['database'].upper()}';
                """
                st.write(f"执行查询: {query}")
                try:
                    tables_df = session.sql(query).to_pandas()
                    st.write("Schema中的表:", tables_df)

                    # 选择表并查看数据
                    if not tables_df.empty:
                        table_name = st.selectbox("选择要查看的表", tables_df['TABLE_NAME'])
                        if table_name:
                            table_query = f"SELECT * FROM {st.session_state['database']}.{st.session_state['schema']}.{table_name} LIMIT 10"
                            st.write(f"执行查询: {table_query}")

                            table_df = session.sql(table_query).to_pandas()
                            st.write(f"{table_name}中的数据:", table_df)
                    else:
                        st.info("当前Schema中没有表")
                except Exception as e:
                    st.error(f"执行查询失败: {e}")
                    st.write("调试信息: 无法获取表名或表数据。")
                finally:
                    session.close()
            else:
                st.error("无法创建Snowflake会话")
        except Exception as e:
            st.error(f"使用Snowflake会话时出错: {e}")
    else:
        st.warning("你尚未登录，请先登录。")

if __name__ == "__main__":
    # 侧边栏凭证输入
    st.sidebar.header("Snowflake凭证")
    user = st.sidebar.text_input("用户名")
    password = st.sidebar.text_input("密码", type="password")
    account = st.sidebar.text_input("账号")
    warehouse = st.sidebar.text_input("仓库")
    database = st.sidebar.text_input("数据库")
    schema = st.sidebar.text_input("Schema")

    if st.sidebar.button("连接到Snowflake"):
        if user and password and account and warehouse and database and schema:
            st.session_state['user'] = user
            st.session_state['password'] = password
            st.session_state['account'] = account
            st.session_state['warehouse'] = warehouse
            st.session_state['database'] = database
            st.session_state['schema'] = schema
            show_data_page()
        else:
            st.warning("请填写所有Snowflake凭证")
    else:
        if 'user' in st.session_state:
            show_data_page()
        else:
            st.warning("请输入你的Snowflake凭证并点击连接按钮。")

代码修改说明 #

1. 会话创建逻辑：移除对get_active_session()的依赖，改用用户输入的凭证创建独立会话，确保运行时使用用户自身的权限上下文。
2. 查询优化：修改表查询语句，明确指定数据库和Schema，避免因会话默认上下文不一致导致的错误。
3. 资源释放：添加finally块确保会话始终被关闭，避免资源泄漏。
4. 空表处理：增加空表场景的提示，提升用户体验。

额外注意事项 #

• 在Snowflake Native Apps中使用用户凭证创建会话时，需确保应用的网络策略允许出站连接到Snowflake服务（默认允许，自定义策略需检查）。
• 如果需求是让应用角色访问固定数据库，给应用角色授予对应权限是正确且永久的方案，示例授权语句：

  GRANT USAGE ON DATABASE YOUR_DB TO ROLE YOUR_APP_RUN_ROLE;
  GRANT USAGE ON SCHEMA YOUR_DB.YOUR_SCHEMA TO ROLE YOUR_APP_RUN_ROLE;
  GRANT SELECT ON ALL TABLES IN SCHEMA YOUR_DB.YOUR_SCHEMA TO ROLE YOUR_APP_RUN_ROLE;
  -- 如需访问未来创建的表，添加以下语句
  GRANT SELECT ON FUTURE TABLES IN SCHEMA YOUR_DB.YOUR_SCHEMA TO ROLE YOUR_APP_RUN_ROLE;
  

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