"""

Streamlit login/register using streamlit-authenticator with SQLite backend.

Credentials are loaded from the users table on each run; new registrations

are persisted via database.db_manager.create_user.

"""

from __future__ import annotations

from typing import Any

import streamlit as st

import streamlit_authenticator as stauth

import config

from database import db_manager

from database.models.users import UserRecord

from utils.encryption import ensure_development_key_file

def build_credentials_from_db() -> dict[str, Any]:

"""

Build the credentials dict expected by streamlit-authenticator from SQLite.

Structure:

{

"usernames": {

"jane": {"email": "...", "name": "...", "password": "<bcrypt>"},

}

}

"""

users = db_manager.list_active_users()

usernames: dict[str, dict[str, str]] = {}

for user in users:

usernames[user.username] = user.to_credentials_entry()

return {"usernames": usernames}

class AuthHandler:

"""

Wraps streamlit-authenticator and syncs state to st.session_state.

Usage in a Streamlit app::

auth = AuthHandler()

auth.render_login()

if not auth.is_authenticated():

st.stop()

"""

def __init__(self) -> None:

if config.IS_DEVELOPMENT:

ensure_development_key_file()

self._credentials = build_credentials_from_db()

self._authenticator = stauth.Authenticate(

self._credentials,

config.AUTH_COOKIE_NAME,

config.AUTH_COOKIE_KEY,

config.AUTH_COOKIE_EXPIRY_DAYS,

)

@property

def authenticator(self) -> stauth.Authenticate:

return self._authenticator

def reload_credentials(self) -> None:

"""Refresh credentials from DB (call after registration)."""

self._credentials = build_credentials_from_db()

self._authenticator.credentials = self._credentials

def render_login(self, location: str = "main", key: str = "Login") -> tuple[str | None, bool | None, str | None]:

"""

Render the login widget.

Returns (name, authentication_status, username) from streamlit-authenticator.

"""

fields = self._authenticator.login(location=location, key=key)

if fields is None:

return None, None, None

name, status, username = fields

if status:

user = db_manager.get_user_by_username(username or "")

if user:

st.session_state["user_id"] = user.user_id

st.session_state["username"] = user.username

st.session_state["user_name"] = user.name

st.session_state["user_email"] = user.email

return name, status, username

def render_logout(self, button_name: str = "Logout", location: str = "sidebar", key: str = "Logout") -> None:

"""Render logout and clear session keys."""

self._authenticator.logout(button_name, location, key=key)

for key_name in ("user_id", "username", "user_name", "user_email"):

st.session_state.pop(key_name, None)

def is_authenticated(self) -> bool:

return st.session_state.get("authentication_status") is True

def get_logged_in_user(self) -> UserRecord | None:

user_id = st.session_state.get("user_id")

if user_id is None:

username = st.session_state.get("username")

if username:

return db_manager.get_user_by_username(username)

return None

return db_manager.get_user_by_id(int(user_id))

def render_register(

self,

*,

location: str = "main",

pre_authorized: list[str] | None = None,

key: str = "Register",

captcha: bool = False,

) -> bool:

"""

Render registration form; on success, persist user to SQLite.

pre_authorized:

None — open registration (first install / development)

list — only listed emails may register (production)

Returns True if a new user was registered and saved to the database.

"""

# Empty list blocks everyone; use None for open registration.

email, username, full_name = self._authenticator.register_user(

location=location,

pre_authorized=pre_authorized,

key=key,

captcha=captcha,

)

if not username:

return False

if db_manager.username_exists(username):

self.reload_credentials()

return False

creds = self._authenticator.credentials.get("usernames", {})

entry = creds.get(username)

if not entry:

return False

db_manager.create_user(

username=username,

email=email or entry.get("email", f"{username}@local"),

name=full_name or entry.get("name", username),

password_hash=entry["password"],

)

self.reload_credentials()

return True

def hash_password(plain_password: str) -> str:

"""Hash a password with bcrypt for storage (streamlit-authenticator compatible)."""

return stauth.Hasher.hash(plain_password)