hansi/co_app
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

added users administration tool

Browse Source
This commit is contained in:
knedlik
2025-12-15 08:21:18 +01:00
parent e2091ec677
commit 13d747827d
11 changed files with 602 additions and 76 deletions
Download Patch File Download Diff File Expand all files Collapse all files

86
app/admin_users.py Normal file
View File

@@ -0,0 +1,86 @@
# admin_users.py
import streamlit as st
import pandas as pd
from db_users import init_db, list_users, get_user, create_user, update_user, set_password, delete_user
st.set_page_config(page_title="User Admin", layout="wide")
init_db()
st.title("Benutzerverwaltung")
# --- Übersicht ---
users = list_users()
df = pd.DataFrame(users)
if df.empty:
st.info("Noch keine Benutzer vorhanden.")
df = pd.DataFrame(columns=["id","username","display_name","email","role","is_active","created_at"])
st.subheader("Übersicht")
st.dataframe(df, use_container_width=True, hide_index=True)
st.divider()
colA, colB = st.columns(2, gap="large")
# --- Neuen Benutzer anlegen ---
with colA:
st.subheader("Neuen Benutzer anlegen")
with st.form("create_user", clear_on_submit=True):
username = st.text_input("Username*", max_chars=50)
password = st.text_input("Initiales Passwort*", type="password")
display_name = st.text_input("Anzeigename")
email = st.text_input("E-Mail")
role = st.selectbox("Rolle", ["user", "admin"], index=0)
is_active = st.checkbox("Aktiv", value=True)
submitted = st.form_submit_button("Anlegen")
if submitted:
if not username or not password:
st.error("Username und Passwort sind Pflicht.")
else:
try:
create_user(username, password, display_name, email, role, is_active)
st.success("Benutzer angelegt.")
st.rerun()
except Exception as e:
st.error(f"Fehler beim Anlegen: {e}")
# --- Bestehenden Benutzer bearbeiten/löschen ---
with colB:
st.subheader("Benutzer bearbeiten / löschen")
user_ids = df["id"].tolist() if "id" in df.columns else []
selected_id = st.selectbox("Benutzer wählen", user_ids, format_func=lambda uid: f"{uid} – {df.loc[df.id==uid,'username'].values[0] if len(df)>0 else uid}" if uid in user_ids else str(uid))
user = get_user(int(selected_id)) if selected_id else None
if user:
with st.form("edit_user"):
display_name2 = st.text_input("Anzeigename", value=user.get("display_name") or "")
email2 = st.text_input("E-Mail", value=user.get("email") or "")
role2 = st.selectbox("Rolle", ["user", "admin"], index=0 if user["role"]=="user" else 1)
is_active2 = st.checkbox("Aktiv", value=bool(user["is_active"]))
save = st.form_submit_button("Änderungen speichern")
if save:
update_user(int(selected_id), display_name2, email2, role2, is_active2)
st.success("Gespeichert.")
st.rerun()
st.markdown("**Passwort zurücksetzen**")
with st.form("reset_pw", clear_on_submit=True):
new_pw = st.text_input("Neues Passwort", type="password")
reset = st.form_submit_button("Passwort setzen")
if reset:
if not new_pw:
st.error("Bitte ein Passwort eingeben.")
else:
set_password(int(selected_id), new_pw)
st.success("Passwort aktualisiert.")
st.markdown("**Löschen**")
confirm = st.checkbox("Ich möchte diesen Benutzer wirklich löschen.")
if st.button("Benutzer löschen", disabled=not confirm):
delete_user(int(selected_id))
st.success("Benutzer gelöscht.")
st.rerun()

BIN
app/app_db/app.db
View File

Binary file not shown.

20
app/app_db/app_db.py
View File

@@ -6,6 +6,11 @@ BASE_DIR = Path(__file__).resolve().parent
# DB_PATH = BASE_DIR / "app_db" / "app.db"
DB_PATH = BASE_DIR / "app.db"
#-------------------------------------------------------------------------------------
# Allgemeine Datenbankfunktionen
#-------------------------------------------------------------------------------------
def get_conn():
# check_same_thread=False, damit Streamlit mehrere Threads nutzen kann
return sqlite3.connect(DB_PATH, check_same_thread=False)
@@ -16,3 +21,18 @@ def get_list(sql, params=None):
conn.close()
return df
def send_cmd(sql, params=None):
try:
with get_conn() as conn:
if params is None:
conn.execute(sql)
else:
conn.execute(sql, params)
return True
except Exception as e:
print("DB-Fehler:", e)
return False
#-------------------------------------------------------------------------------------
#
#-------------------------------------------------------------------------------------

73
app/db_users.py Normal file
View File

@@ -0,0 +1,73 @@
# db_users.py
import sqlite3
from contextlib import contextmanager
from typing import Optional
import bcrypt
DB_PATH = "app.db"
@contextmanager
def get_conn():
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
finally:
conn.close()
def init_db():
with get_conn() as conn:
conn.execute("""
CREATE TABLE IF NOT EXISTS users (
id INTEGER PRIMARY KEY AUTOINCREMENT,
username TEXT NOT NULL UNIQUE,
display_name TEXT,
email TEXT,
password_hash BLOB NOT NULL,
role TEXT NOT NULL DEFAULT 'user',
is_active INTEGER NOT NULL DEFAULT 1,
created_at TEXT NOT NULL DEFAULT (datetime('now'))
)
""")
def list_users():
with get_conn() as conn:
rows = conn.execute("""
SELECT id, username, display_name, email, role, is_active, created_at
FROM users ORDER BY id
""").fetchall()
return [dict(r) for r in rows]
def get_user(user_id: int) -> Optional[dict]:
with get_conn() as conn:
r = conn.execute("""
SELECT id, username, display_name, email, role, is_active
FROM users WHERE id = ?
""", (user_id,)).fetchone()
return dict(r) if r else None
def create_user(username: str, password: str, display_name: str = "", email: str = "", role: str = "user", is_active: bool = True):
pw_hash = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())
with get_conn() as conn:
conn.execute("""
INSERT INTO users (username, display_name, email, password_hash, role, is_active)
VALUES (?, ?, ?, ?, ?, ?)
""", (username, display_name, email, pw_hash, role, 1 if is_active else 0))
def update_user(user_id: int, display_name: str, email: str, role: str, is_active: bool):
with get_conn() as conn:
conn.execute("""
UPDATE users
SET display_name=?, email=?, role=?, is_active=?
WHERE id=?
""", (display_name, email, role, 1 if is_active else 0, user_id))
def set_password(user_id: int, new_password: str):
pw_hash = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt())
with get_conn() as conn:
conn.execute("UPDATE users SET password_hash=? WHERE id=?", (pw_hash, user_id))
def delete_user(user_id: int):
with get_conn() as conn:
conn.execute("DELETE FROM users WHERE id=?", (user_id,))

41
app/db_users_search.py Normal file
View File

@@ -0,0 +1,41 @@
# db_users_search.py
import sqlite3
from contextlib import contextmanager
DB_PATH = "app.db"
@contextmanager
def get_conn():
conn = sqlite3.connect(DB_PATH, check_same_thread=False)
conn.row_factory = sqlite3.Row
try:
yield conn
conn.commit()
finally:
conn.close()
def search_users(q: str, limit: int, offset: int):
q = (q or "").strip()
like = f"%{q}%"
with get_conn() as conn:
rows = conn.execute("""
SELECT id, username, display_name, email, role, is_active, created_at
FROM users
WHERE (? = '')
OR (username LIKE ? COLLATE NOCASE
OR display_name LIKE ? COLLATE NOCASE
OR email LIKE ? COLLATE NOCASE)
ORDER BY username
LIMIT ? OFFSET ?
""", (q, like, like, like, limit, offset)).fetchall()
total = conn.execute("""
SELECT COUNT(*)
FROM users
WHERE (? = '')
OR (username LIKE ? COLLATE NOCASE
OR display_name LIKE ? COLLATE NOCASE
OR email LIKE ? COLLATE NOCASE)
""", (q, like, like, like)).fetchone()[0]
return [dict(r) for r in rows], int(total)

BIN
app/images/home.png Normal file
View File

Binary file not shown.
Side by Side

After

Width:  |  Height:  |  Size: 308 B

2
app/pages/home.py
View File

@@ -8,7 +8,7 @@ from tools.load_css import load_css
load_css()
st.set_page_config(page_title="Co-App Home", page_icon="🏠")
st.set_page_config(page_title="Co-App Home", page_icon=":material/home:", layout="centered", initial_sidebar_state="expanded")
authenticator = require_login()
st.session_state["authenticator"] = authenticator

427
app/pages/user.py
View File

@@ -1,56 +1,399 @@
# import streamlit as st
# import pandas as pd
# from db_users import init_db, get_user, update_user, set_password # wie zuvor
# from db_users_search import search_users # wie zuvor
# st.set_page_config(layout="wide")
# init_db()
# if "selected_user_id" not in st.session_state:
# st.session_state.selected_user_id = None
# def open_edit(user_id: int):
# st.session_state.selected_user_id = user_id
# st.session_state.show_edit = True
# @st.dialog("Benutzer bearbeiten")
# def edit_user_dialog(user_id: int):
# user = get_user(user_id)
# if not user:
# st.error("Benutzer nicht gefunden.")
# return
# st.markdown(f"**{user['username']}** (ID {user['id']})")
# with st.form("edit_user_form"):
# display_name = st.text_input("Anzeigename", value=user.get("display_name") or "")
# email = st.text_input("E-Mail", value=user.get("email") or "")
# role = st.selectbox("Rolle", ["user", "admin"], index=0 if user["role"] == "user" else 1)
# is_active = st.checkbox("Aktiv", value=bool(user["is_active"]))
# save = st.form_submit_button("Speichern")
# if save:
# update_user(user_id, display_name, email, role, is_active)
# st.success("Gespeichert.")
# st.session_state.show_edit = False
# st.rerun()
# st.divider()
# st.markdown("**Passwort zurücksetzen**")
# with st.form("pw_form", clear_on_submit=True):
# pw = st.text_input("Neues Passwort", type="password")
# pw_save = st.form_submit_button("Passwort setzen")
# if pw_save:
# if not pw:
# st.error("Bitte Passwort eingeben.")
# else:
# set_password(user_id, pw)
# st.success("Passwort aktualisiert.")
# st.session_state.show_edit = False
# st.rerun()
# st.title("Benutzerverwaltung")
# q = st.text_input("Suche", placeholder="username / Name / E-Mail")
# rows, total = search_users(q, limit=20, offset=0)
# st.caption(f"{total} Treffer")
# for r in rows:
# c1, c2, c3, c4 = st.columns([2, 2, 3, 1])
# c1.write(r["username"])
# c2.write(r.get("display_name") or "")
# c3.write(r.get("email") or "")
# if c4.button("Ändern", key=f"edit_{r['id']}"):
# open_edit(r["id"])
# # Dialog öffnen, wenn Flag gesetzt
# if st.session_state.get("show_edit") and st.session_state.selected_user_id:
# edit_user_dialog(int(st.session_state.selected_user_id))
# # admin_users.py
# import streamlit as st
# import pandas as pd
# from db_users import init_db, list_users, get_user, create_user, update_user, set_password, delete_user
# st.set_page_config(page_title="User Admin", layout="wide")
# init_db()
# st.title("Benutzerverwaltung")
# # --- Übersicht ---
# users = list_users()
# df = pd.DataFrame(users)
# if df.empty:
# st.info("Noch keine Benutzer vorhanden.")
# df = pd.DataFrame(columns=["id","username","display_name","email","role","is_active","created_at"])
# st.subheader("Übersicht")
# st.dataframe(df, use_container_width=True, hide_index=True)
# st.divider()
# colA, colB = st.columns(2, gap="large")
# # --- Neuen Benutzer anlegen ---
# with colA:
# st.subheader("Neuen Benutzer anlegen")
# with st.form("create_user", clear_on_submit=True):
# username = st.text_input("Username*", max_chars=50)
# password = st.text_input("Initiales Passwort*", type="password")
# display_name = st.text_input("Anzeigename")
# email = st.text_input("E-Mail")
# role = st.selectbox("Rolle", ["user", "admin"], index=0)
# is_active = st.checkbox("Aktiv", value=True)
# submitted = st.form_submit_button("Anlegen")
# if submitted:
# if not username or not password:
# st.error("Username und Passwort sind Pflicht.")
# else:
# try:
# create_user(username, password, display_name, email, role, is_active)
# st.success("Benutzer angelegt.")
# st.rerun()
# except Exception as e:
# st.error(f"Fehler beim Anlegen: {e}")
# # --- Bestehenden Benutzer bearbeiten/löschen ---
# with colB:
# st.subheader("Benutzer bearbeiten / löschen")
# user_ids = df["id"].tolist() if "id" in df.columns else []
# selected_id = st.selectbox("Benutzer wählen", user_ids, format_func=lambda uid: f"{uid} – {df.loc[df.id==uid,'username'].values[0] if len(df)>0 else uid}" if uid in user_ids else str(uid))
# user = get_user(int(selected_id)) if selected_id else None
# if user:
# with st.form("edit_user"):
# display_name2 = st.text_input("Anzeigename", value=user.get("display_name") or "")
# email2 = st.text_input("E-Mail", value=user.get("email") or "")
# role2 = st.selectbox("Rolle", ["user", "admin"], index=0 if user["role"]=="user" else 1)
# is_active2 = st.checkbox("Aktiv", value=bool(user["is_active"]))
# save = st.form_submit_button("Änderungen speichern")
# if save:
# update_user(int(selected_id), display_name2, email2, role2, is_active2)
# st.success("Gespeichert.")
# st.rerun()
# st.markdown("**Passwort zurücksetzen**")
# with st.form("reset_pw", clear_on_submit=True):
# new_pw = st.text_input("Neues Passwort", type="password")
# reset = st.form_submit_button("Passwort setzen")
# if reset:
# if not new_pw:
# st.error("Bitte ein Passwort eingeben.")
# else:
# set_password(int(selected_id), new_pw)
# st.success("Passwort aktualisiert.")
# st.markdown("**Löschen**")
# confirm = st.checkbox("Ich möchte diesen Benutzer wirklich löschen.")
# if st.button("Benutzer löschen", disabled=not confirm):
# delete_user(int(selected_id))
# st.success("Benutzer gelöscht.")
# st.rerun()
import streamlit as st
from auth_runtime import require_login
from ui.sidebar import build_sidebar
from auth import create_user
from pathlib import Path
from tools.load_css import load_css
from app_db.app_db import get_list, send_cmd
from ui.selectboxes import get_roles, get_id
DASH_NAME = Path(__file__).stem # Hier muss die dash_id aus der DB stehen -> wird gegen die session_state geprüft (User-Berechtigung)
load_css()
st.set_page_config(page_title="Co-App Benutzer", page_icon="🏠")
st.set_page_config(page_title="Co-App Benutzer", page_icon=":material/person:", layout="wide", initial_sidebar_state="collapsed")
authenticator = require_login()
st.session_state["authenticator"] = authenticator
def sidebar():
fullname = st.session_state.get("fullname")
role_text = st.session_state.get("role_text")
with st.sidebar:
st.logo("app/images/GMN_Logo_neu_rgb.png", size="small")
st.markdown(f"**{fullname}** ({role_text})")
col1, col2, col3 = st.columns([2,2,1])
# with col1:
# if st.button("Logout", use_container_width=True):
# st.rerun()
col1, col2 = st.columns([2,2])
with col1:
authenticator.logout("Logout")
# st.rerun()
# if st.button("Home", use_container_width=True):
# st.switch_page("pages/home.py")
with col2:
if st.button("🏠 Home", use_container_width=True):
if st.button("Home", use_container_width=True, icon=":material/home:"):
st.switch_page("pages/home.py")
user()
def user():
@st.dialog("Benutzer anlegen")
def dialog_create_user():
txt_username = st.text_input("Benutzername")
txt_firstname = st.text_input("Vorname")
txt_lastname = st.text_input("Nachname")
txt_email = st.text_input("Email")
txt_pwd = st.text_input("Kennwort", type="password")
cmb_role = st.selectbox("Rolle", get_roles(), index=None)
if st.button("Save"):
if create_user(
username=txt_username,
firstname=txt_firstname,
lastname=txt_lastname,
email=txt_email,
role_id=get_id(cmb_role),
password=txt_pwd
):
st.session_state.save_msg = f"✅ Benutzer '{txt_username}' erfolgreich gespeichert"
else:
st.session_state.save_msg = "❌ Fehler beim Speichern"
st.rerun()
@st.dialog("Benutzer löschen")
def dialog_delete_user(id):
if id == None:
st.write("kein Benutzer ausgewählt")
else:
df = get_list("select username from users where id = ?",(id,))
username = df.iloc[0]["username"]
st.write(f"Der Benutzer {username} wird gelöscht! Sind Sie sicher?")
if st.button("Löschen"):
if username != "admin":
if send_cmd("delete from users where id = ?",(id,)):
st.session_state.delete_msg = f"✅ Benutzer '{username}' erfolgreich gelöscht!"
else:
st.session_state.delete_msg = f"❌ Benutzer '{username}' konnte nicht gelöscht werden!"
else:
st.session_state.delete_msg = f"❌ Benutzer '{username}' darf nicht gelöscht werden!"
st.rerun()
@st.dialog("Benutzer bearbeiten")
def dialog_modify_user(id):
if id == None:
st.write("kein Benutzer ausgewählt")
else:
sql = """
select
u.id,
u.username,
u.firstname,
u.lastname,
u.role_id || ' | ' || r.role_text as role,
r.role_text,
u.new_pwd,
u.active
from
users u
left join roles r
on u.role_id = r.role_id
where u.id = ?
"""
df = get_list(sql,(id,))
# df = get_list("select username from users where id = ?",(id,))
st.header("Benutzerverwaltung")
st.session_state.orig_user_data = df
df_roles = get_roles()
roles = df_roles["text"].tolist()
role = df.iloc[0]["role"]
try:
idx = roles.index(role)
except:
idx = None
txt_username = st.text_input("Benutzername", value=df.iloc[0]["username"])
txt_firstname = st.text_input("Vorname",df.iloc[0]["firstname"])
txt_lastname = st.text_input("Nachname",df.iloc[0]["username"])
txt_email = st.text_input("Email", df.iloc[0]["username"])
txt_pwd = st.text_input("Passwort", placeholder="Neues Passwort eingeben", type="password")
new_pwd = st.checkbox("Neues Passwort",df.iloc[0]["new_pwd"])
cmb_role = st.selectbox("Rolle", roles, placeholder="Rolle auswählen", index=idx)
# if st.button("Save"):
# if create_user(
# username=txt_username,
# firstname=txt_firstname,
# lastname=txt_lastname,
# email=txt_email,
# role_id=get_id(cmb_role),
# password=txt_pwd
# ):
# st.session_state.save_msg = f"✅ Benutzer '{txt_username}' erfolgreich gespeichert"
# else:
# st.session_state.save_msg = "❌ Fehler beim Speichern"
# st.rerun()
def user():
if "selected_user_id" not in st.session_state:
st.session_state.selected_user_id = None
tab_user, tab_role, tab_permission = st.tabs(["Benutzer", "Rollen", "Berechtigungen"])
#--------------------------------------------------------------------------------------------------
# Benutzerverwaltung
#--------------------------------------------------------------------------------------------------
with tab_user:
df = get_list("""
select
u.id,
u.username,
u.firstname,
u.lastname,
u.role_id || ' | ' || r.role_text as role,
u.new_pwd,
u.active
from
users u
left join roles r
on u.role_id = r.role_id
""")
col_find_user, col_create_user, col_modify_user, col_delete_user = st.columns([3,2,2,2], vertical_alignment="bottom")
with col_find_user:
txt_search = st.text_input("", placeholder="Benutzer, Vorname, ...", icon=":material/search:")
with col_create_user:
if st.button(label="Benutzer anlegen", use_container_width=True, icon=":material/person_add:"):
dialog_create_user()
if "save_msg" in st.session_state:
st.toast(st.session_state.save_msg)
del st.session_state.save_msg
with col_modify_user:
if st.button(label="Benutzer bearbeiten", use_container_width=True, icon=":material/person:"):
dialog_modify_user(st.session_state.selected_user_id)
with col_delete_user:
if st.button(label="Benutzer löschen", use_container_width=True, icon=":material/person_remove:"):
dialog_delete_user(st.session_state.selected_user_id)
if "delete_msg" in st.session_state:
st.toast(st.session_state.delete_msg)
del st.session_state.delete_msg
if txt_search.strip():
txt_search_norm = txt_search.strip().lower()
mask = (
df["username"].fillna("").str.lower().str.contains(txt_search_norm))
df_view = df.loc[mask].copy()
else:
df_view = df.copy()
event = st.dataframe(df_view,key="data", on_select="rerun", selection_mode="single-row")
rows = event.selection.rows
if rows:
row_idx = rows[0]
st.session_state.selected_user_id = int(df_view.iloc[row_idx]["id"])
else:
st.session_state.selected_user_id = None
if __name__ == "__main__":
sidebar()
@@ -63,53 +406,3 @@ if __name__ == "__main__":
# username = st.session_state.get("username")
# df = st.session_state.get("df_sidebar")
# if check(df,DASH_NAME) == False:
# st.markdown("**FEHLER**")
# st.error("Die Seite kann nicht angezeigt werden - keine Berechtigung!")
# st.stop()
# df = st.session_state.get("df_sidebar")
# st.text(DASH_NAME)
# st.text(st.session_state.get(username))
# print(df)
# st.title("Benutzerverwaltung")
# with st.expander("Neuen Nutzer anlegen"):
# new_u = st.text_input("Neuer Username", key="new_u")
# new_fname = st.text_input("Vorname", key="new_fname")
# new_lname = st.text_input("Nachname", key="new_lname")
# new_email = st.text_input("E-Mail", key="new_email")
# new_p = st.text_input("Neues Passwort", type="password", key="new_p")
# new_role = st.selectbox("Rolle", ["user", "admin"], key="new_role")
# if st.button("Anlegen"):
# if new_u and new_p:
# ok = create_user(
# new_u.strip(),
# new_p,
# new_role,
# new_email.strip() or None,
# new_fname.strip() or None,
# new_lname.strip() or None,
# )
# st.success("Nutzer angelegt.") if ok else st.error(
# "Username bereits vorhanden oder Fehler."
# )
# else:
# st.warning("Bitte Username und Passwort eingeben.")
# st.subheader("Dein Bereich")
# st.write(f"Personalisierter Content für **{username}**.")

19
app/ui/selectboxes.py Normal file
View File

@@ -0,0 +1,19 @@
from app_db.app_db import get_list
def get_roles():
sql = """
select
role_id || ' | ' || role_text as text
from
roles
"""
df = get_list(sql)
return df
def get_id(id_text: str):
id = int(id_text.split("|")[0])
if not id:
id = ""
return id

10
app/ui/sidebar.py
View File

@@ -7,9 +7,6 @@ from app_db.app_db import get_conn, get_list
def build_sidebar():
# if st.session_state.get("authentication_status") != True:
# return
authenticator = st.session_state.get("authenticator")
username = st.session_state.get("name")
role_text = st.session_state.get("role_text")
@@ -24,14 +21,14 @@ def build_sidebar():
st.logo("app/images/GMN_Logo_neu_rgb.png", size="small")
st.markdown(f"**{fullname}** ({role_text})")
col1, col2, col3 = st.columns([2,2,1])
col1, col2 = st.columns([2,2])
with col1:
authenticator.logout("Logout")
with col2:
if st.button("Home", use_container_width=True):
if st.button("Home", use_container_width=True, icon=":material/home:"):
st.switch_page("pages/home.py")
# --- Suchfeld ---
@@ -41,9 +38,6 @@ def build_sidebar():
# Aktive Seite ermitteln (für Expander-Status / Highlight)
current_page = st.session_state.get("_page_path")
# --- DF filtern, falls Suchbegriff gesetzt ---
if query:
mask = (