JiriUhlir
126 lines
2.9 KiB
Python
126 lines
2.9 KiB
Python
from typing import Any
|
|
|
|
from app.db.database import get_connection
|
|
from app.db.migrations import run_migrations
|
|
|
|
ROLES = ("admin", "developer", "viewer")
|
|
|
|
|
|
def _table_columns(con, table_name: str) -> set[str]:
|
|
rows = con.execute(f"PRAGMA table_info({table_name})").fetchall()
|
|
return {row["name"] for row in rows}
|
|
|
|
|
|
def list_users() -> list[dict[str, Any]]:
|
|
run_migrations()
|
|
con = get_connection()
|
|
columns = _table_columns(con, "users")
|
|
last_login_select = "last_login_at" if "last_login_at" in columns else "NULL AS last_login_at"
|
|
|
|
rows = con.execute(
|
|
f"""
|
|
SELECT
|
|
id,
|
|
username,
|
|
display_name,
|
|
email,
|
|
role,
|
|
COALESCE(is_enabled, 1) AS is_enabled,
|
|
created_at,
|
|
updated_at,
|
|
{last_login_select}
|
|
FROM users
|
|
ORDER BY id
|
|
"""
|
|
).fetchall()
|
|
|
|
con.close()
|
|
return [dict(row) for row in rows]
|
|
|
|
|
|
def get_user(user_id: int) -> dict[str, Any] | None:
|
|
run_migrations()
|
|
con = get_connection()
|
|
columns = _table_columns(con, "users")
|
|
last_login_select = "last_login_at" if "last_login_at" in columns else "NULL AS last_login_at"
|
|
|
|
row = con.execute(
|
|
f"""
|
|
SELECT
|
|
id,
|
|
username,
|
|
display_name,
|
|
email,
|
|
role,
|
|
COALESCE(is_enabled, 1) AS is_enabled,
|
|
created_at,
|
|
updated_at,
|
|
{last_login_select}
|
|
FROM users
|
|
WHERE id = ?
|
|
""",
|
|
(user_id,),
|
|
).fetchone()
|
|
|
|
con.close()
|
|
return dict(row) if row else None
|
|
|
|
|
|
def count_enabled_admins(excluding_user_id: int | None = None) -> int:
|
|
run_migrations()
|
|
con = get_connection()
|
|
params: list[Any] = []
|
|
exclusion = ""
|
|
if excluding_user_id is not None:
|
|
exclusion = "AND id != ?"
|
|
params.append(excluding_user_id)
|
|
|
|
row = con.execute(
|
|
f"""
|
|
SELECT COUNT(*) AS count
|
|
FROM users
|
|
WHERE LOWER(COALESCE(role, '')) = 'admin'
|
|
AND COALESCE(is_enabled, 1) = 1
|
|
{exclusion}
|
|
""",
|
|
params,
|
|
).fetchone()
|
|
|
|
con.close()
|
|
return int(row["count"] or 0)
|
|
|
|
|
|
def update_user_role(user_id: int, role: str) -> None:
|
|
if role not in ROLES:
|
|
raise ValueError("Invalid user role")
|
|
|
|
run_migrations()
|
|
con = get_connection()
|
|
con.execute(
|
|
"""
|
|
UPDATE users
|
|
SET role = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(role, user_id),
|
|
)
|
|
con.commit()
|
|
con.close()
|
|
|
|
|
|
def set_user_enabled(user_id: int, enabled: bool) -> None:
|
|
run_migrations()
|
|
con = get_connection()
|
|
con.execute(
|
|
"""
|
|
UPDATE users
|
|
SET is_enabled = ?,
|
|
updated_at = CURRENT_TIMESTAMP
|
|
WHERE id = ?
|
|
""",
|
|
(1 if enabled else 0, user_id),
|
|
)
|
|
con.commit()
|
|
con.close()
|