retoor
ca6c527e32
The primary administrator was the earliest Admin by created_at with no further condition, so a soft-deleted or deactivated account could hold the role and then be refused by the api-key path, leaving nobody able to use the database API, the backup download or cross-owner container management. Rows with no recorded signup time also sorted ahead of every real account. Scan the earliest admins instead and take the first that is neither deleted nor deactivated, with missing timestamps sorted last. Also stop the projects listing returning 500 when project_type or description is NULL (the dict default never applies to an existing NULL column), align the issue test fixture with its unit twin so a combined run cannot collide on a fixed uid, read the settings value from the database rather than a stale per-process cache, and give the seeded and fixture admins the is_active and created_at fields that every real signup writes.
124 lines
3.7 KiB
Python
124 lines
3.7 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
from .core import TTLCache, bump_cache_version, db, sync_local_cache
|
|
|
|
|
|
def get_users_by_uids(uids):
|
|
if not uids or "users" not in db.tables:
|
|
return {}
|
|
users = db["users"]
|
|
if "uid" not in users.columns:
|
|
return {}
|
|
seen = set()
|
|
unique = [u for u in uids if u not in seen and not seen.add(u)]
|
|
return {u["uid"]: u for u in users.find(users.table.columns.uid.in_(unique))}
|
|
|
|
|
|
_admins_cache = TTLCache(ttl=300, max_size=4)
|
|
# The primary administrator must be an account that can actually authenticate, so scan a
|
|
# few of the earliest admins and skip any that are soft-deleted or deactivated.
|
|
PRIMARY_ADMIN_CANDIDATES = 50
|
|
|
|
|
|
def invalidate_admins_cache() -> None:
|
|
_admins_cache.clear()
|
|
bump_cache_version("admins")
|
|
|
|
|
|
def get_admin_uids():
|
|
sync_local_cache("admins", _admins_cache)
|
|
cached = _admins_cache.get("uids")
|
|
if cached is not None:
|
|
return list(cached)
|
|
if "users" not in db.tables:
|
|
return []
|
|
rows = db.query("SELECT uid FROM users WHERE role = 'Admin'")
|
|
uids = [row["uid"] for row in rows]
|
|
_admins_cache.set("uids", uids)
|
|
return list(uids)
|
|
|
|
|
|
def set_user_timezone(user_uid: str, tz_name: str) -> None:
|
|
if "users" not in db.tables or not user_uid or not tz_name:
|
|
return
|
|
users = db["users"]
|
|
if not users.has_column("timezone"):
|
|
users.create_column_by_example("timezone", "")
|
|
current = users.find_one(uid=user_uid)
|
|
if current and current.get("timezone") == tz_name:
|
|
return
|
|
users.update({"uid": user_uid, "timezone": tz_name}, ["uid"])
|
|
|
|
|
|
def set_last_seen(user_uid: str, iso: str) -> None:
|
|
if "users" not in db.tables or not user_uid or not iso:
|
|
return
|
|
users = db["users"]
|
|
if not users.has_column("last_seen"):
|
|
users.create_column_by_example("last_seen", "")
|
|
users.update({"uid": user_uid, "last_seen": iso}, ["uid"])
|
|
|
|
|
|
def get_online_users(cutoff_iso: str, limit: int = 30) -> list:
|
|
if "users" not in db.tables:
|
|
return []
|
|
users = db["users"]
|
|
if "last_seen" not in users.columns:
|
|
return []
|
|
return list(
|
|
users.find(
|
|
last_seen={">=": cutoff_iso},
|
|
order_by=["username"],
|
|
_limit=limit,
|
|
)
|
|
)
|
|
|
|
|
|
def _can_hold_primary_admin(row, tracks_active):
|
|
if row.get("deleted_at"):
|
|
return False
|
|
return not tracks_active or bool(row.get("is_active"))
|
|
|
|
|
|
def get_primary_admin_uid():
|
|
sync_local_cache("admins", _admins_cache)
|
|
cached = _admins_cache.get("primary")
|
|
if cached is not None:
|
|
return cached or None
|
|
if "users" not in db.tables:
|
|
return None
|
|
rows = list(
|
|
db.query(
|
|
"SELECT * FROM users WHERE role = 'Admin' "
|
|
"ORDER BY (created_at IS NULL OR created_at = ''), created_at ASC, id ASC "
|
|
"LIMIT :cap",
|
|
cap=PRIMARY_ADMIN_CANDIDATES,
|
|
)
|
|
)
|
|
tracks_active = "is_active" in db["users"].columns
|
|
primary = next(
|
|
(row["uid"] for row in rows if _can_hold_primary_admin(row, tracks_active)),
|
|
None,
|
|
)
|
|
_admins_cache.set("primary", primary or "")
|
|
return primary
|
|
|
|
|
|
def search_users_by_username(q, *, exclude_uid=None, limit=10):
|
|
if not q or "users" not in db.tables:
|
|
return []
|
|
if exclude_uid is not None:
|
|
rows = db.query(
|
|
"SELECT uid, username FROM users WHERE username LIKE :q AND uid != :me LIMIT :limit",
|
|
q=f"%{q}%",
|
|
me=exclude_uid,
|
|
limit=limit,
|
|
)
|
|
else:
|
|
rows = db.query(
|
|
"SELECT uid, username FROM users WHERE username LIKE :q LIMIT :limit",
|
|
q=f"%{q}%",
|
|
limit=limit,
|
|
)
|
|
return [{"uid": r["uid"], "username": r["username"]} for r in rows]
|