使用 Snowflake (SiS) 中的 Streamlit 為每個使用者進行個人化

介紹

2024 年 7 月底，Snowflake 中的 Streamlit 中提供了 Current_User 和行存取策略。

這個更新令人興奮的部分是，現在可以輕鬆、安全地識別連接到應用程式的 Snowflake 用戶並為每個用戶自訂處理。

無需設定自訂登入機製或使用者管理表，您可以透過以下方式為每個使用者個性化單一應用程式：

• 更改每個使用者的應用程式顯示
• 為每位使用者準備個人化分析儀表板
• 使用行存取策略為每位使用者取得不同的查詢輸出結果（企業版以上版本）

在這篇文章中，我們將建立一個簡單的待辦事項清單來顯示個人使用者資訊。

註：本文僅代表個人觀點，不代表Snowflake。

功能概述

目標

• 使用單一共用應用程式管理個人待辦事項清單
• 使用行存取策略來阻止顯示其他人的待辦事項

完成的影像

使用者TKANNO的螢幕

使用者TARO的螢幕

先決條件

• 雪花帳戶
  • 需要企業版帳戶才能使用行存取政策

筆記

• Snowflake 中的 Streamlit 以擁有者權限運行，因此 Current_Role 將與 Snowflake 中的 Streamlit 應用程式角色相同。 （因此不能用於個人化）

程式

建立一個表格來儲存 ToDo 列表

從工作表執行以下指令：

-- Create ToDo list table
CREATE TABLE IF NOT EXISTS todo_list (
    id INT AUTOINCREMENT,
    task VARCHAR(255),
    status VARCHAR(20),
    due_date DATE,
    completed_date DATE,
    owner VARCHAR(50)
);

建立行存取策略

此策略傳回 todo_list 表中的擁有者與連接到 Snowflake 應用程式中的 Streamlit 的 current_user 相符的行。

從工作表執行以下指令：

-- Create row access policy
CREATE ROW ACCESS POLICY IF NOT EXISTS todo_row_access_policy
    AS (owner VARCHAR) RETURNS BOOLEAN ->
        owner = CURRENT_USER();

應用程式存取策略

從工作表執行以下指令：

-- Apply row access policy
ALTER TABLE todo_list ADD ROW ACCESS POLICY todo_row_access_policy ON (owner);

至此完成了工作表操作。

在 Snowflake 應用程式中運行 Streamlit

在 Snowflake 應用程式中建立一個新的 Streamlit，然後複製並貼上以下程式碼：

第 14 行是以字串形式檢索連接到應用程式的目前使用者的位置。

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

# Layout settings
st.set_page_config(
    layout="wide"
)

# Get Snowflake session
session = get_active_session()

# Get current user
current_user = session.sql("SELECT CURRENT_USER()").collect()[0][0]

# Get ToDo list
def get_todo_list():
    return session.table("todo_list").to_pandas()

# Add or update task
def upsert_task(task_id, task, status, due_date, completed_date):
    due_date_sql = f"'{due_date}'" if due_date else "NULL"
    completed_date_sql = f"'{completed_date}'" if completed_date else "NULL"

    if task_id:
        session.sql(f"""
        UPDATE todo_list
        SET task = '{task}', status = '{status}', due_date = {due_date_sql}, completed_date = {completed_date_sql}
        WHERE id = {task_id}
        """).collect()
    else:
        session.sql(f"""
        INSERT INTO todo_list (task, status, owner, due_date, completed_date)
        VALUES ('{task}', '{status}', '{current_user}', {due_date_sql}, {completed_date_sql})
        """).collect()

# Delete task
def delete_task(task_id):
    session.sql(f"DELETE FROM todo_list WHERE id = {task_id}").collect()

# Main function
def main():
    st.title(f"{current_user}'s Personal Dashboard")

    # Task list
    st.subheader(f"{current_user}'s ToDo List")
    todo_df = get_todo_list()

    # Display header
    col1, col2, col3, col4, col5 = st.columns([3, 2, 2, 2, 2])
    col1.write("Task")
    col2.write("Status")
    col3.write("Due Date")
    col4.write("Completed Date")
    col5.write("Delete")

    # Display task list
    for _, row in todo_df.iterrows():
        col1, col2, col3, col4, col5 = st.columns([3, 2, 2, 2, 2])

        with col1:
            task = st.text_input("task", value=row['TASK'], key=f"task_{row['ID']}", label_visibility="collapsed")

        with col2:
            status = st.selectbox("status", ["Pending", "In Progress", "Completed"], index=["Pending", "In Progress", "Completed"].index(row['STATUS']), key=f"status_{row['ID']}", label_visibility="collapsed")

        with col3:
            due_date = st.date_input("due_date", value=pd.to_datetime(row['DUE_DATE']).date() if pd.notna(row['DUE_DATE']) else None, key=f"due_date_{row['ID']}", label_visibility="collapsed")

        with col4:
            completed_date = st.date_input("comp_date", value=pd.to_datetime(row['COMPLETED_DATE']).date() if pd.notna(row['COMPLETED_DATE']) else None, key=f"completed_date_{row['ID']}", label_visibility="collapsed")

        with col5:
            if st.button("Delete", key=f"delete_{row['ID']}"):
                delete_task(row['ID'])
                st.experimental_rerun()

        # Update database immediately if values change
        if task != row['TASK'] or status != row['STATUS'] or due_date != row['DUE_DATE'] or completed_date != row['COMPLETED_DATE']:
            upsert_task(row['ID'], task, status, due_date, completed_date)
            st.experimental_rerun()

    # Add new task
    st.subheader("Add New Task")
    new_task = st.text_input("New Task")
    new_status = st.selectbox("Status", ["Pending", "In Progress", "Completed"])
    new_due_date = st.date_input("Due Date")
    if st.button("Add"):
        upsert_task(None, new_task, new_status, new_due_date, None)
        st.success("New task added")
        st.experimental_rerun()

# Main process
if __name__ == "__main__":
    main()

結論

你覺得怎麼樣？透過結合 Current_User 和行存取策略，您可以透過簡單的步驟為每個使用者建立個人化的安全應用程式。這為根據您的想法創建更用戶友好的應用程式提供了可能性。

一些先進的想法包括在 Snowflake 中透過 Streamlit 寫入表時添加 Current_User 資訊作為簽名，或使用個人化資訊作為 Cortex LLM 的上下文來建立個人助理。

請嘗試用 Current_User 的有趣用途來挑戰自己！

公告

Snowflake 最新動態 X 更新

我正在分享 Snowflake 在 X 上的最新動態。如果您有興趣，請隨時關注！

英文版

雪花新鮮事機器人（英文版）
https://x.com/snow_new_en

日文版

雪花What's New Bot（日文版）
https://x.com/snow_new_jp

變更歷史記錄

(20240914) 初始貼文

日本原創文章

https://zenn.dev/tsubasa_tech/articles/a23029dfe97c46

以上是使用 Snowflake (SiS) 中的 Streamlit 為每個使用者進行個人化的詳細內容。更多資訊請關注PHP中文網其他相關文章！