【Streamlit v1.58.0】ログインユーザーと個別のファイルとを紐付ける
remy_world複数人でログインできる環境下で、個別データを簡単なExcelファイルで管理している場合、他人のExcelファイルにアクセスして欲しくありません。その場合、ログインユーザーと紐付いたファイル構成にしないといけません。それにはログインユーザー専用のフォルダを作成して、そのフォルダ内のファイルしかアクセスできないようにすれば簡単に出来るのでは?と考えてました。ファイルは何でも良いのですが、ここではCSVファイルを例にした実装方法について考えました。
ログイン認証は『【Streamlit v1.46.1】認証モジュール + SQLite3を使ってログイン機能を実装』で説明した方法を使います。そして複数人のユーザーが利用する環境下を想定し、ログインしたユーザー毎に自分専用のフォルダを持ち、そのフォルダ内のファイルしかアクセス出来ないようなシステムを組んでみます。一方アドミニの管理者のみ全ファイルにアクセスできるようにします。
（※)前回よりもStreamlitのversionは、v1.46.1からv1.58.0に上がっていました。
目次
SQLiteのテーブル変更(ユーザー管理)
- usersテーブルにroleカラムを追加し、管理者の'admin'か、通常の'member'か区別する。
- 登録時はデフォルトでmemberとする。
createdb_table.py
import sqlite3
import logging
# 接続先のデータベースが存在しない場合、新たに作成してから接続します。
conn = sqlite3.connect("st-members.db")
c = conn.cursor()
# usersテーブルの作成
try:
c.execute("drop table if exists users")
c.execute("""
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL,
password TEXT NOT NULL,
role TEXT default('member')
)
""")
except Exception as e:
logging.exception("エラーが発生しました: %s", e)
else :
print("データベースの作成に成功しました")このファイルを実行(python createdb_table.py)して、データベースとテーブルを作成します。
フォルダ構成
adminならusera_data内の全ファイルをアクセスでき、一般ユーザー(’member')は各人のフォルダ内のファイルのみアクセス可能とします。
またadminはadmin専用のファイルを持っているとし、以下のフォルダ構成で検討します。
project/
├── app.py
├── users_data/
│ ├── 1/ # id=1のusernameがsanjiとすると
│ │ ├── sanji_日付1.csv
│ │ └── sanji_日付2.csv
│ │ └── ...
│ ├── 2/ # id=2のusernameがamanoとすると
│ │ └── amano_日付3.csv
│ │ └── amano_日付4.csv
│ │ └── ...
│ └── admin/
│ └── 管理者用_日付5.csv
└── st_members.db # データベース各人のフォルダは別途カラムを追加しても良いのですが、ここでは簡単のためユニークなid番号で代用します。
ファイルの閲覧
Streamlitはセキュリティ上の理由からOSのフォルダ選択ダイアログを直接提供していません。その代わり、os.listdir() などを使ってサーバー上のディレクトリパスを読み取り、st.selectbox でフォルダやファイルを選択させるUIを実装するのが一般的です。
selectboxの注意点
Streamlitでは何かUIが変わるたびに全ページを再描画(rerun())します。selectboxの戻り値をそのまま使うとrerun()関数で消えてしまうので必ずsession_stateに保存する必要があります。
# ▼ 初期化（初回のみ）
if "mypage" not in st.session_state:
st.session_state["mypage"] = 'Home'
# ----------------------
# selectboxで遷移先選択
# ----------------------
dash_page = st.selectbox('選択してください',['Home','お知らせ', 'お問い合わせ', 'ファイル開く'],key="sidebar_page")
st.session_state["mypage"] = dash_pageこれはログイン後に画面選択させるコードです。選択結果のdash_pageをそのまま使わずに、session_stateに保存しています。
selsectbox使い方のポイント
忘れないように、ポイントとしてまとめておきます。
- 選択値を session_state に保存する。
- selectbox には必ず key を付ける。
key がない selectbox は Streamlit が『同じ UI』とみなし、前回の選択を勝手に保持するため、選択が反映されない。 - key が衝突すると選択が保持されないので必ず別名にする。
- index=None を使わない。index=None を使わず、session_state を初期値にする。
index=None では『未選択(None)』のまま rerun() されるため、選択が反映されない。
個人専用ファイルを閲覧するコード
ログイン後の画面選択で'ファイル開く'画面に遷移したときのみ、ファイル閲覧できるようにしています。この場合roleの権限で閲覧できる範囲を変えたいので、if文で分岐させています。閲覧するファイルはここではcsvファイルに限定します。
import streamlit as st
import os
from pathlib import Path
# ▼ 初期化（初回のみ）
if "myselected_files" not in st.session_state:
st.session_state["myselected_files"] = []
DATA_DIR = "users_data"
# -------------------------
# ファイル開くの選択画面
# -------------------------
def csvfile_edit():
st.header("ファイル開く")
user_name=st.session_state['name']
row = get_prop(user_name)
data_dir = row["id"] # ←idをデイレクトリ名としているため
role = row["role"]
csv_page(data_dir, role)
# -----------------------------
# ログインユーザーの権限取得
# -----------------------------
def get_prop(username):
conn = sqlite3.connect("st-members.db")
cur=conn.cursor()
cur.execute("select id, role from users where username = ? ",(username,))
row = cur.fetchone()
if row:
return {"id": str(row[0]), "role":row[1]}
return None
# ---------------------------
# CSV 閲覧ページ
# ---------------------------
def csv_page(dir, role):
options = None
# -----
# 管理者は全フォルダを閲覧可能
# -----
if role == "admin":
# admin権限なので一番上のデイレクトリパスを指定
dir_path = Path(DATA_DIR) # users_data
# 🔽指定したフォルダの中を再帰的（下層のフォルダもすべて）検索。
# 下記はcsvファイルのみ検索します。
# ①
file_paths = [p for p in dir_path.rglob("*.csv") if p.is_file()]
# 🔽keyがfile_pathsの文字列、valueがファイル名となる辞書形式で取り出しています。
# ②
file_dict = {str(p): p.name for p in file_paths}
options = list(file_dict.values())
# セレクトボックスを作成:ファイル名で選択
selected_file = st.selectbox(
"ファイルを選択してください",
options=options,
key="admin_option",
# 🔽 ③
index=options.index(st.session_state["myselected_files"]) if st.session_state["myselected_files"] in options else 0,
placeholder="ファイルを選択してください..."
)
# 🔽 選択したファイルをsession_stateに保存
st.session_state["myselected_files"] = selected_file
# -----
# member会員は自分のフォルダ内のみ閲覧可能。
# -----
else:
dir_path = Path(DATA_DIR + "/" + dir)
os.makedirs(dir_path, exist_ok=True)
file_paths = [p for p in dir_path.rglob("*.csv") if p.is_file()]
file_dict = {str(p): p.name for p in file_paths} # ←ファイル名
options = list(file_dict.values())
# セレクトボックスを作成:ファイル名で選択する
selected_file = st.selectbox(
"検索するファイルを選択してください",
options=options,
key="member_option",
index=options.index(st.session_state["myselected_files"]) if st.session_state["myselected_files"] in options else 0,
placeholder="ファイルを選択してください..."
)
# ▼ 選択したファイル
st.session_state["myselected_files"] = selected_file
簡単な説明（コード中①〜③）
①file_paths = [p for p in dir_path.rglob("*.csv") if p.is_file()]
rglob() は、指定したフォルダの中を再帰的（下層のフォルダもすべて）に探索し、条件に合うファイルやフォルダのパスをまとめて取得する標準ライブラリpathlib の便利な機能です。
引数に"*.csv"渡すことでcsvファイルのみを検索できます。
②file_dict = {str(p): p.name for p in file_paths}
取得したパスオブジェクトから、keyに文字列、valueがファイル名となる辞書形式で取り出しています。p.nameはオブジェクトから「ファイル名だけ」を取り出します。
③index=options.index(st.session_state["myselected_files"])
if st.session_state["myselected_files"] in options else 0,
これは、index=None を使わず、session_state を初期値にするコードです。
selectboxのindexはリストの番号（0から始まる整数）を指定して設定します。
session_stateに保存されている選択値がoptions に含まれていれば、その値のindexを使う。含まれていなければindex=0（先頭）を使うという意味です。
因みに、リスト型のoptionsの.index(値)は、指定した値が最初に出てくる位置を返すものです。
アプリに組み込む
方針としてはサイドバーは状態更新のみ、描画はメイン画面としています。ファイル選択もログイン後のサイドバーで行っても良いのですが、常に表示される事になるので、少し邪魔な感じがしました。なので、画面選択で"ファイル開く"を選択したときのみ、そのメイン画面でファイルが選択できるようにしました。
app.py
全体のコードは以下の通りです。
# v1.58.0
import streamlit as st
# ↓画像を扱いたい方は以下もimportすると便利です。
from PIL import Image
import sqlite3
import os
import streamlit_authenticator as stauth # 認証
import pandas as pd
from pathlib import Path
DATA_DIR = "users_data"
# -------------------------
# 状態初期化
# -------------------------
def init_state():
if "state" not in st.session_state:
st.session_state["state"] = "wait"
if "mypage" not in st.session_state:
st.session_state["mypage"] = "Home"
if "myselected_files" not in st.session_state:
st.session_state["myselected_files"] = []
# -------------------------
# ログイン後のメイン遷移
# main画面
# -------------------------
def Home():
st.header('ホーム画面')
img1="./images/公園の河.jpg"
img2="./images/palazzina寺院.jpg"
col1,col2 = st.columns(2)
with col1:
image1= Image.open(img1)
st.image(image1)
with col2:
image2 = Image.open(img2)
st.image(image2)
def info():
st.header('お知らせ')
st.write("〇〇公園のひまわりが見頃です。")
st.write("〇〇町の祇園祭が無事終了しました。")
def inquiry():
st.header('お問い合わせ')
st.text_input("氏名")
st.text_input("連絡先")
# --------------------------------------------
# Main側でここを開いたときだけ選択画面が表れる
# --------------------------------------------
def csvfile_edit():
st.header("ファイル開く")
user_name=st.session_state['name']
row = get_prop(user_name)
# print(row["id"]+ " : " +row["role"])
data_dir = row["id"]
role = row["role"]
csv_page(data_dir, role)
# --------------------------------------------------
# ログインユーザーの権限取得(ログイン時のみ)
# --------------------------------------------------
def get_prop(username):
conn = sqlite3.connect("st-members.db")
cur=conn.cursor()
cur.execute("select id, role from users where username = ? ",(username,))
row = cur.fetchone()
if row:
return {"id": str(row[0]), "role":row[1]}
return None
# --------------------------------------------------------------------
# CSV 閲覧ページ
# ---------------------------------------------------------------------
def csv_page(dir, role):
options = None
# 管理者は全フォルダを閲覧可能
if role == "admin":
dir_path = Path(DATA_DIR)
# 再帰的にすべてのファイルを取得（Path.rglobを使用）
# サブフォルダも含めて検索したい拡張子を指定
file_paths = [p for p in dir_path.rglob("*.csv") if p.is_file()]
file_dict = {str(p): p.name for p in file_paths}
options = list(file_dict.values())
# セレクトボックスを作成:ファイル名で選択する
selected_file = st.selectbox(
"ファイルを選択してください",
options=options,
key="admin_option",
index=options.index(st.session_state["myselected_files"]) if st.session_state["myselected_files"] in options else 0,
placeholder="ファイルを選択してください..."
)
st.session_state["myselected_files"] = selected_file
else:
dir_path = Path(DATA_DIR + "/" + dir)
os.makedirs(dir_path, exist_ok=True)
file_paths = [p for p in dir_path.rglob("*.csv") if p.is_file()]
file_dict = {str(p): p.name for p in file_paths} # ←ファイル名
options = list(file_dict.values())
# セレクトボックスを作成:ファイル名で選択する
selected_file = st.selectbox(
"検索するファイルを選択してください",
options=options,
key="member_option",
index=options.index(st.session_state["myselected_files"]) if st.session_state["myselected_files"] in options else 0,
placeholder="ファイルを選択してください..."
)
# ▼ 選択したファイル
st.session_state["myselected_files"] = selected_file
# -------------------------
# credentials作成
# -------------------------
def fetch_users(conn):
try:
cur = conn.cursor()
cur.execute("select username ,password from users")
rows = cur.fetchall()
credentials = {"usernames":{}}
for row in rows:
username, hashed_password = row
tmp_dict = {
'name': username,
'password': hashed_password,
}
credentials["usernames"][username] =tmp_dict
conn.close()
return credentials
except Exception as e:
st.error(f"ユーザー情報取得エラー: {e}")
return {"usernames": {}}
# -------------------------
# 新規メンバー登録
# -------------------------
def regist_user(conn, username, hashed_password):
conn = sqlite3.connect("st-members.db")
try:
cur = conn.cursor()
cur.execute("insert into users (username, password) values (?, ?)",(username, hashed_password))
conn.commit()
conn.close()
return True
except sqlite3.IntegrityError :
st.warning("このユーザー名は既に登録されています。別のユーザー名をお試しください。")
return False
except Exception as e:
st.error(f"ユーザー登録エラー: {e}")
return False
# -------------------------
# メイン関数
# -------------------------
def main():
conn = sqlite3.connect("st-members.db")
if conn is None:
st.error("データベースへの接続に失敗しました。アプリケーションを終了します。")
st.stop() # データベース接続ができない場合はここで停止
mycredentials = fetch_users(conn)
# ↓本番環境ではしてはいけない。
mysignature_key = 'super_secret34567815!#$_verylonglong-hot'
authenticator = stauth.Authenticate(
mycredentials,
'streamlit_v1.47_login_cookie', # 任意のクッキー名
mysignature_key, # 任意の署名キー
cookie_expiry_days=30, # クッキーの有効期限（日数）
)
# サイドバーの設計
with st.sidebar:
# 状態初期化
init_state()
# 待機状態
if st.session_state["state"]== "wait":
if st.button("新規登録はこちらから"):
st.session_state["state"] = "signin_req"
st.rerun()
st.divider()
# st.header('ログイン')
authenticator.login(location="sidebar",fields={
"Form name": "ログイン",
"Username": "ユーザー名",
"Password": "パスワード",
"Login": "ログイン"
})
# if auth_status:
if st.session_state["authentication_status"] :
st.session_state["state"] = "logined"
elif st.session_state["authentication_status"] is False:
st.error("ユーザー名またはパスワードが間違っています")
elif st.session_state["authentication_status"] is None:
st.warning("ユーザー名とパスワードを入力してください")
# ログイン成功
elif st.session_state["state"]== "logined":
if st.session_state["authentication_status"]:
st.write("ログインに成功しました。")
st.success(f"ようこそ {st.session_state['name']} さん！")
# ----------------------
# selectboxで遷移先選択
# ----------------------
dash_page = st.selectbox('選択してください',['Home','お知らせ', 'お問い合わせ', 'ファイル開く'],key="sidebar_page")
st.session_state["mypage"] = dash_page
if st.session_state["authentication_status"] is True :
authenticator.logout("ログアウト", location="sidebar")
if st.session_state["authentication_status"] is None :
st.session_state["state"] = "wait"
st.warning("ログアウトしました。再度ログインしてください。")
st.rerun()
# 新規登録
elif st.session_state["state"] == "signin_req":
st.header('登録フォーム')
st.write("アカウントをお持ちでない方はこちらから登録してください。")
with st.form("register_user_form"):
username = st.text_input("ユーザー名")
new_password = st.text_input("パスワード", type="password")
confirm_password = st.text_input("パスワードの確認", type="password")
submit_registration = st.form_submit_button("登録する")
if submit_registration:
if username and new_password and confirm_password:
if new_password == confirm_password :
try:
hashed_password = stauth.Hasher.hash(new_password)
if regist_user(conn, username, hashed_password):
st.session_state["state"] = "wait"
st.markdown("<span style='color:blue;'>ご登録ありがとうございました。</span>", unsafe_allow_html=True)
st.rerun()
except Exception as e:
st.error(f"パスワードのハッシュ化または登録中にエラーが発生しました: {e}")
else:
st.error("パスワードが一致しません。再度確認してください。")
else :
st.error("全ての登録項目を入力してください。")
st.divider()
if st.button("ログイン画面に戻る"):
st.session_state["state"] = "wait"
st.rerun()
# -------------------------
# メイン画面の切り替え処理
# -------------------------
if st.session_state["state"] == "logined":
page = st.session_state.get("mypage", "Home")
# 画面遷移
if page == "Home":
Home()
elif page == "お知らせ":
info()
elif page == "お問い合わせ":
inquiry()
elif page == "ファイル開く":
csvfile_edit()
if __name__ == '__main__':
main()
アプリの動作確認
動作確認
ファイル構成：3名のmemberとadminの1名で動作確認
id=5を新規登録、新規登録も正常に動作しています。
アプリの動画
以上です😀
Follow me!
