Snowflake (SiS) の Streamlit を使用してユーザーごとにパーソナライズ

導入

2024 年 7 月末に、Current_User および Row Access Policy が Snowflake の Streamlit で利用可能になりました。

このアップデートの興味深い点は、アプリケーションに接続している Snowflake ユーザーを簡単かつ安全に識別し、各ユーザーの処理をカスタマイズできることです。

カスタム ログイン メカニズムやユーザー管理テーブルをセットアップする必要がなく、次のような方法でユーザーごとに 1 つのアプリケーションをパーソナライズできます。

• ユーザーごとにアプリの表示を変更する
• ユーザーごとにパーソナライズされた分析ダッシュボードを準備する
• 行アクセス ポリシーを使用してユーザーごとに異なるクエリ出力結果を取得する (Enterprise Edition 以降)

この投稿では、個々のユーザー情報を表示するシンプルな ToDo リストを作成します。

注: この投稿は私の個人的な見解を表すものであり、Snowflake の見解ではありません。

機能の概要

目標

• 単一の共有アプリを使用して個人の ToDo リストを管理
• 行アクセス ポリシーを使用して、他の人の ToDo が表示されないようにします

完成イメージ

ユーザー TKANO の画面

ユーザーTAROの画面

前提条件

• スノーフレークアカウント
  • 行アクセス ポリシーを使用するには Enterprise Edition アカウントが必要です

注記

• 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 の興味深い使い方にぜひ挑戦してみてください!

お知らせ

X の Snowflake の新機能アップデート

X に関する Snowflake の新着情報を共有しています。ご興味がございましたら、お気軽にフォローしてください。

英語版

Snowflake 新着情報ボット (英語版)
https://x.com/snow_new_ja

日本語版

Snowflake 新着情報ボット (日本語版)
https://x.com/snow_new_jp

変更履歴

(20240914) 最初の投稿

日本語オリジナル記事

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

以上がSnowflake (SiS) の Streamlit を使用してユーザーごとにパーソナライズの詳細内容です。詳細については、PHP 中国語 Web サイトの他の関連記事を参照してください。