Snowflake上Streamlit应用获取用户仓库列表的问题求助

问题根源 #

1. 原生Streamlit应用仅支持创建/使用OWNER'S RIGHTS模式的存储过程，无法启用EXECUTE AS CALLER，存储过程执行时的会话上下文为应用所有者，而非调用用户。
2. 原方案中LAST_QUERY_ID()依赖前序SHOW WAREHOUSES的查询上下文，但在OWNER权限模式下，两次查询的会话上下文不匹配，导致出现query_id_place_holder_XYZ not found错误。

解决方案：直接查询系统视图替代SHOW+RESULT_SCAN #

放弃依赖SHOW命令和RESULT_SCAN的方式，改用Snowflake内置的INFORMATION_SCHEMA.WAREHOUSES视图获取仓库列表，该方式更稳定且符合原生应用的权限限制。

1. 修改存储过程定义（setup.sql）

CREATE OR REPLACE PROCEDURE public.get_available_warehouses()
RETURNS TABLE(
    name VARCHAR,
    type VARCHAR,
    size VARCHAR,
    state VARCHAR,
    is_default BOOLEAN,
    is_current BOOLEAN,
    auto_suspend NUMBER,
    auto_resume BOOLEAN,
    min_cluster_count NUMBER,
    max_cluster_count NUMBER,
    scaling_policy VARCHAR,
    resource_monitor VARCHAR,
    comment VARCHAR
)
LANGUAGE PYTHON
RUNTIME_VERSION = '3.8'
PACKAGES = ('snowflake-snowpark-python')
HANDLER = 'get_available_warehouses'
OWNER'S RIGHTS
AS
$$
def get_available_warehouses(session):
    # 直接查询INFORMATION_SCHEMA获取当前账户下的仓库列表
    return session.sql("""
        SELECT 
            WAREHOUSE_NAME AS name,
            WAREHOUSE_TYPE AS type,
            WAREHOUSE_SIZE AS size,
            WAREHOUSE_STATE AS state,
            IS_DEFAULT AS is_default,
            IS_CURRENT AS is_current,
            AUTO_SUSPEND AS auto_suspend,
            AUTO_RESUME AS auto_resume,
            MIN_CLUSTER_COUNT AS min_cluster_count,
            MAX_CLUSTER_COUNT AS max_cluster_count,
            SCALING_POLICY AS scaling_policy,
            RESOURCE_MONITOR AS resource_monitor,
            COMMENT AS comment
        FROM INFORMATION_SCHEMA.WAREHOUSES
        WHERE CURRENT_ACCOUNT() = WAREHOUSE_OWNER
    """)
$$;

2. Streamlit应用调用代码

import snowflake.snowpark as snowpark
import streamlit as st

def main(session: snowpark.Session):
    # 调用存储过程并转换为Pandas DataFrame
    warehouses_df = session.sql("CALL public.get_available_warehouses()").to_pandas()
    
    # 生成下拉菜单选项
    if not warehouses_df.empty:
        warehouse_options = warehouses_df['NAME'].tolist()
        selected_warehouse = st.selectbox("选择仓库", warehouse_options)
        st.success(f"已选择仓库：{selected_warehouse}")
    else:
        st.warning("未找到可用仓库")

if __name__ == "__main__":
    session = snowpark.Session.builder.appName("WarehouseSelector").getOrCreate()
    main(session)

进阶权限过滤（可选） #

如果需要返回当前用户有权限使用的仓库（而非整个账户下的所有仓库），可修改存储过程中的SQL查询为：

SELECT w.WAREHOUSE_NAME AS name
FROM INFORMATION_SCHEMA.WAREHOUSES w
JOIN INFORMATION_SCHEMA.GRANTS_TO_USERS g
    ON w.WAREHOUSE_NAME = g.GRANTEE_NAME
WHERE g.GRANTEE_NAME = CURRENT_USER()
    AND g.PRIVILEGE IN ('USAGE', 'MODIFY')
GROUP BY w.WAREHOUSE_NAME

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