概要
Snowflake のタスクの実行プログラムを外部ファイル化する方法の記事で，ストアドプロシージャやタスクの権限がないロールのユーザに対して，ストアドプロシージャを外部ファイル化することで，擬似的にその権限がないユーザでも，任意のSQLまたはPythonプログラムに編集可能な仕組みを紹介した．一方で，本仕組みの場合，システムとユーザの責任の分離に関する課題がある点についても紹介した．
今回は，ストアドプロシージャやタスクの権限がないロールのユーザに対して，ストアドプロシージャを編集可能な仕組みをStreamlitにてアプリとして提供することを考える．この仕組みの特徴として，編集した任意のSQLをユーザ側で動作確認を行えるため，Streamlitのアプリの作成はシステム側の責任で行い，Streamlitのアプリを経由して実行するSQLの動作確認はユーザ側の責任で行えるため，SQLの編集以降はユーザ側の責任とすることができる．今回は，タスクは事前に準備している想定で，ストアドプロシージャの中にあるSQL部分のみを編集する形としているが，タスク部分もユーザ側で編集可能な作りとすることで，完全にシステムとユーザの責任を分離することが可能である．
サンプルコード
import snowflake.connector
import streamlit as st
import pandas as pd
# Snowflake接続設定
def get_snowflake_connection():
return snowflake.connector.connect(
user="<USERNAME>",
password="<PASSWORD>",
account="<ACCOUNT>",
database="<DATABASE_NAME>",
schema="<SCHEMA_NAME>",
role="<ROLE_NAME>",
warehouse="<WAREHOUSE_NAME>"
)
# ストアドプロシージャを作成する関数
def create_stored_procedure(proc_name, sql_code):
try:
conn = get_snowflake_connection()
cursor = conn.cursor()
# ストアドプロシージャのSQL
procedure_sql = f"""
CREATE OR REPLACE PROCEDURE {proc_name}()
RETURNS STRING
LANGUAGE SQL
EXECUTE AS CALLER
AS $$
BEGIN
{sql_code}
RETURN 'Procedure executed successfully';
END;
$$;
"""
cursor.execute(procedure_sql)
return "ストアドプロシージャが正常に作成されました！"
except Exception as e:
return f"エラーが発生しました: {e}"
finally:
cursor.close()
conn.close()
# ストアドプロシージャを実行する関数
def execute_stored_procedure(proc_name):
try:
conn = get_snowflake_connection()
cursor = conn.cursor()
# ストアドプロシージャの実行
exec_sql = f"CALL {proc_name}();"
cursor.execute(exec_sql)
result = cursor.fetchone() # 結果を取得
return f"ストアドプロシージャの実行結果: {result[0]}"
except Exception as e:
return f"エラーが発生しました: {e}"
finally:
cursor.close()
conn.close()
# Streamlit UI
st.title("Snowflakeストアドプロシージャ作成＆実行ツール")
# ユーザー入力
proc_name = st.text_input("ストアドプロシージャ名", placeholder="例: my_stored_procedure")
sql_code = st.text_area("ストアドプロシージャ内のSQLを入力", height=200, placeholder="例: SELECT * FROM my_table;")
# ストアドプロシージャ作成
if st.button("作成"):
if not proc_name.strip():
st.error("ストアドプロシージャ名を入力してください。")
elif not sql_code.strip():
st.error("SQLコードを入力してください。")
else:
result = create_stored_procedure(proc_name.strip(), sql_code.strip())
if "エラー" in result:
st.error(result)
else:
st.success(result)
# ストアドプロシージャ実行
if st.button("実行"):
if not proc_name.strip():
st.error("ストアドプロシージャ名を入力してください。")
else:
result = execute_stored_procedure(proc_name.strip())
if "エラー" in result:
st.error(result)
else:
st.success(result)
応用（セキュリティ強化）
サンプルコードで実装すると，ストアドプロシージャやタスクの権限がないロールのユーザに対して，ストアドプロシージャを編集可能な状態を実現できるが，ストアドプロシージャやタスクの権限を持ったユーザでの実行と成るため，本アプリを経由することで，本来ユーザができないSQLまで実行できてしまう可能性が残る．そのため，その点を解消するために，2点の制限を加えることで，ユーザのロールの範囲内でのストアドプロシージャを編集しかできないようにした．
- ストアドプロシージャ内のSQLの先頭に，use role <ROLE_NAME>を記載することで，そのユーザが実行できないSQLは抑止する．Snowflake側の仕様により，use roleによるロールの変更は1つのSQL内で1度しか反映されないため，ユーザ側でuse roleを追加してもロールの変更はできないことを利用している．
- ストアドプロシージャ名の入力ボックスをなくし，proc_name = "<DATABASE_NAME>.<SCHEMA_NAME>.<PROCEDURE_NAME>"に変更することでストアドプロシージャ名をユーザ側で指定させないように抑止する
import snowflake.connector
import streamlit as st
import pandas as pd
# Snowflake接続設定
def get_snowflake_connection():
return snowflake.connector.connect(
user="<USERNAME>",
password="<PASSWORD>",
account="<ACCOUNT>",
database="<DATABASE_NAME>",
schema="<SCHEMA_NAME>",
role="<ROLE_NAME>",
warehouse="<WAREHOUSE_NAME>"
)
# ストアドプロシージャを作成する関数
def create_stored_procedure(proc_name, sql_code):
try:
conn = get_snowflake_connection()
cursor = conn.cursor()
# ストアドプロシージャのSQL
procedure_sql = f"""
CREATE OR REPLACE PROCEDURE {proc_name}()
RETURNS STRING
LANGUAGE SQL
EXECUTE AS CALLER
AS $$
BEGIN
use role <ROLE_NAME>
{sql_code}
RETURN 'Procedure executed successfully';
END;
$$;
"""
cursor.execute(procedure_sql)
return "ストアドプロシージャが正常に作成されました！"
except Exception as e:
return f"エラーが発生しました: {e}"
finally:
cursor.close()
conn.close()
# ストアドプロシージャを実行する関数
def execute_stored_procedure(proc_name):
try:
conn = get_snowflake_connection()
cursor = conn.cursor()
# ストアドプロシージャの実行
exec_sql = f"CALL {proc_name}();"
cursor.execute(exec_sql)
result = cursor.fetchone() # 結果を取得
return f"ストアドプロシージャの実行結果: {result[0]}"
except Exception as e:
return f"エラーが発生しました: {e}"
finally:
cursor.close()
conn.close()
# Streamlit UI
st.title("Snowflakeストアドプロシージャ作成＆実行ツール")
# ユーザー入力
proc_name = "<DATABASE_NAME>.<SCHEMA_NAME>.<PROCEDURE_NAME>"
sql_code = st.text_area("ストアドプロシージャ内のSQLを入力", height=200, placeholder="例: SELECT * FROM my_table;")
# ストアドプロシージャ作成
if st.button("作成"):
if not proc_name.strip():
st.error("ストアドプロシージャ名を入力してください。")
elif not sql_code.strip():
st.error("SQLコードを入力してください。")
else:
result = create_stored_procedure(proc_name.strip(), sql_code.strip())
if "エラー" in result:
st.error(result)
else:
st.success(result)
# ストアドプロシージャ実行
if st.button("実行"):
if not proc_name.strip():
st.error("ストアドプロシージャ名を入力してください。")
else:
result = execute_stored_procedure(proc_name.strip())
if "エラー" in result:
st.error(result)
else:
st.success(result)
修正
EXECUTE AS OWNERを記載しないと，デフォルトで*EXECUTE AS OWNER
*が追加され，AS $$以降の実行SQLに権限がない場合エラーとなるため，
EXECUTE AS CALLERを明示的に記載する必要がある．
- 変更前
CREATE OR REPLACE PROCEDURE {proc_name}()
RETURNS STRING
LANGUAGE SQL
AS $$
- 変更前（自動適用）
CREATE OR REPLACE PROCEDURE {proc_name}()
RETURNS STRING
LANGUAGE SQL
EXECUTE AS OWNER
AS $$
- 変更後
CREATE OR REPLACE PROCEDURE {proc_name}()
RETURNS STRING
LANGUAGE SQL
EXECUTE AS CALLER
AS $$