【Streamlit v1.58.0】認証モジュールのパスワードの変更
認証モジュールの仕様として「ユーザー名(username)」自体はユニークで同じusernameは使えません。これはユーザー名がデータのキーとして使われているためですが、この重複はシステムが自動で判断して弾いてくれません。
SQLite3と連携して新規ユーザー登録機能を自作する場合は、開発者がコード側で「すでに同じ username が存在するかどうか」を事前にチェックして弾くロジックを実装する必要がありました。(※前回【Streamlit v1.46.1】認証モジュール + SQLite3を使ってログイン機能を実装では、その部分がありません。)
今回は、新規ユーザー登録する前に、usernameの重複が無いか事前にチェックする機能とパスワードの変更機能を実装する場合のメモ書きです。
また、テーブル設計ではusernameを主キーにしないといけないのですが、いかにも重複しそうなカラムを主キーにして、エラーチェックで弾くのはどうかと感じましたので、主キーは変更せず「【Streamlit v1.58.0】ログインユーザーと個別のファイルとを紐付ける」で紹介したテーブル構造のまま実装しています。
目次
usernameの重複を回避
SQLite3のテーブル構造
conn = sqlite3.connect("データベース名")
cur = conn.cursor()
cur.execute("""
CREATE TABLE users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT ,
password TEXT NOT NULL,
role TEXT default('member')
)
""")重複を回避する方法
新規登録を行う際は、データベースに INSERT する前に必ず SELECT 文で既存のユーザーを検索し、存在する場合は画面に警告を表示して処理をストップさせます。
重複を事前に弾くコード例
# -------------------------
# 新規メンバー登録
# -------------------------
def regist_user(conn, username, hashed_password):
conn = sqlite3.connect("st-members.db") # 一旦conn.close()しているので再度接続する!
# 🔽【最重要】同じusernameがすでに存在するかチェック
cur = conn.cursor()
cur.execute("select username from users where username = ?",(username,))
if cur.fetchone():
# データが見つかった(重複している)場合は、エラーを表示して処理を終了
st.error(f"エラー: 『{username}』はすでに使用されています。別のユーザー名をお試しください。")
conn.close()
return False
try:
cur.execute("insert into users (username, password) values (?, ?)",(username, hashed_password))
conn.commit()
conn.close()
return True
except Exception as e:
st.error(f"ユーザー登録エラー: {e}")
return Falseログインパスワードの変更
変更はauthenticator.reset_password()を実行するだけです。引数に渡した mycredentials 辞書の中身は、自動的に新しいハッシュ化されたパスワードに書き換わります。しかしこれは一時的な変更でアプリを再起動すると元に戻ります。必ず(SQLite3やMySQLなどで)永続化しておく必要があります。
パスワードの変更には、セキュリティ対策上の制約条件が自動で追加されます。
エラーコード
- Between 8 and 20 characters long ⇨ 8文字以上20文字以下にすること
- Contain at least one digit ⇨ 数字を含むこと
- Contain at least one lowercase letter ⇨ 小文字を含むこと
- Contain at least one uppercase letter ⇨ 大文字を含むこと
- Contain at least one special character (@$!%*?&) ⇨ 特殊文字を使うこと
変更と永続化のコード例
if authenticator.reset_password(st.session_state["name"]):
st.success("パスワードが正常に変更されました。")
new_hashed_pw = mycredentials["usernames"][st.session_state["name"]]["password"]
# 以下にSQLite3でパスワードを更新します。
update_password_db(user_id,new_hashed_pw)変更するコード例
# -------------------------------------
# パスワードの変更をDBに反映
# -------------------------------------
def update_password_db(id,hashed_password):
conn = sqlite3.connect("st-members.db")
cur = conn.cursor()
cur.execute(
"update users set password = ? where id =?",(hashed_password, id)
)
conn.commit()
conn.close()パスワードの変更をDBに反映させるコード例
まとめ
app.pyの全コードとアプリの動作確認
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
import io
from pathlib import Path # os.pathで出来ることは全てできる
DATA_DIR = "users_data"
DATABASE_NAME = "st-members.db"
# -------------------------
# 状態初期化
# -------------------------
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"
# -------------------------
# ログイン後のメイン遷移
# 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("連絡先")
# -----------------------------------------
# ログインユーザーの権限取得
# usernameと一致する id,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
# ---------------------------------------------
# credentials辞書を作成して返す処理
# ---------------------------------------------
def fetch_users(conn):
try:
cur = conn.cursor()
cur.execute("select username ,password from users")
rows = cur.fetchall()
credentials = {"usernames":{}}
# credentials = {}
for row in rows:
username, hashed_password = row
tmp_dict = {
'name': username, # ここでusernameをnameとして使用
'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") # 一旦conn.close()しているので再度接続する!
# 🔽【最重要】同じusernameがすでに存在するかチェック
cur = conn.cursor()
cur.execute("select username from users where username = ?",(username,))
if cur.fetchone():
# データが見つかった(重複している)場合は、エラーを表示して処理を終了
st.error(f"エラー: 『{username}』はすでに使用されています。別のユーザー名をお試しください。")
conn.close()
return False
try:
cur.execute("insert into users (username, password) values (?, ?)",(username, hashed_password))
conn.commit()
conn.close()
return True
except Exception as e:
st.error(f"ユーザー登録エラー: {e}")
return False
# -------------------------------------
# パスワードの変更をDBに反映
# パスワードをデータベースに上書き保存
# -------------------------------------
def update_password_db(id,hashed_password):
conn = sqlite3.connect("st-members.db")
cur = conn.cursor()
cur.execute(
"update users set password = ? where id =?",(hashed_password, id)
)
conn.commit()
conn.close()
# -------------------------
# メイン関数
# -------------------------
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
# ---------------------------
# パスワード更新 @2026/09/10
# ---------------------------
st.divider()
st.subheader("⚙️ パスワードの変更")
try:
# Between 8 and 20 characters long
# Contain at least one uppercase letter
# Contain at least one special character (@$!%*?&)
old_hashed_pw = mycredentials["usernames"][st.session_state["name"]]["password"]
if authenticator.reset_password(st.session_state["name"]):
st.success("パスワードが正常に変更されました。")
new_hashed_pw = mycredentials["usernames"][st.session_state["name"]]["password"]
user_name = st.session_state["name"]
row = get_prop(user_name)
user_id = row["id"]
update_password_db(user_id,new_hashed_pw)
except Exception as e:
st.error(e)
st.divider()
st.subheader("ログアウトしますか")
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")
# st.header('ダッシュボード')
# 画面遷移
if page == "Home":
Home()
elif page == "お知らせ":
info()
elif page == "お問い合わせ":
inquiry()
if __name__ == '__main__':
main()
アプリの動作確認
①同じusernameでの登録を弾く処理
②ログインパスワードの変更処理
以上です😁
