Snowflake(SiS)의 Streamlit을 사용하여 각 사용자에 맞게 개인화하세요.-파이썬 튜토리얼-php.cn

집

백엔드 개발

파이썬 튜토리얼

Snowflake(SiS)의 Streamlit을 사용하여 각 사용자에 맞게 개인화하세요.

DDD

Sep 14, 2024 pm 12:15 PM

소개

2024년 7월 말에 Current_User 및 행 액세스 정책이 Snowflake의 Streamlit에서 제공되었습니다.

이 업데이트의 흥미로운 점은 이제 애플리케이션에 연결된 Snowflake 사용자를 쉽고 안전하게 식별하고 각 사용자에 대한 처리를 맞춤화할 수 있다는 것입니다.

맞춤형 로그인 메커니즘이나 사용자 관리 테이블을 설정할 필요 없이 다음과 같은 방식으로 각 사용자에 대해 단일 애플리케이션을 개인화할 수 있습니다.

사용자별 애플리케이션 표시 변경
사용자별 맞춤형 분석 대시보드 준비
행 액세스 정책을 사용하여 각 사용자에 대해 서로 다른 쿼리 출력 결과 얻기(Enterprise Edition 이상)

이번 포스팅에서는 개별 사용자 정보를 표시하는 간단한 ToDo 목록을 만들어 보겠습니다.

참고: 이 게시물은 Snowflake의 의견이 아닌 개인적인 견해를 나타냅니다.

기능 개요

목표

단일 공유 앱을 사용하여 개인 ToDo 목록 관리
행 액세스 정책을 사용하여 다른 사람의 할 일이 표시되지 않도록 방지

완성된 이미지

Personalize for each user with Streamlit in Snowflake (SiS)
사용자 TKANNO의 화면

Personalize for each user with Streamlit in Snowflake (SiS)
사용자 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 테이블의 소유자가 Streamlit in Snowflake 애플리케이션에 연결된 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()

로그인 후 복사