ad/molodetz/database/users.py

# retoor <retoor@molodetz.nl>
from molodetz.cache import TTLCache
from molodetz.constants import ROLE_ADMIN, ROLE_MEMBER
from molodetz.database.core import db, generate_uid, in_clause, now_iso
from molodetz.database.settings import bump_cache_version, get_setting, sync_local_cache

_admins_cache = TTLCache(ttl=300, max_size=4)


def users_table():
    return db.get_table("users")


def get_user(uid):
    if not uid or "users" not in db.tables:
        return None
    return db["users"].find_one(uid=uid, deleted_at=None)


def get_user_by_username(username):
    if not username or "users" not in db.tables:
        return None
    return db["users"].find_one(username=username, deleted_at=None)


def get_user_by_email(email):
    if not email or "users" not in db.tables:
        return None
    return db["users"].find_one(email=email.lower(), deleted_at=None)


def get_user_by_api_key(api_key):
    if not api_key or "users" not in db.tables:
        return None
    return db["users"].find_one(api_key=api_key, deleted_at=None)


def users_by_uids(uids):
    uids = [uid for uid in set(uids) if uid]
    if not uids or "users" not in db.tables:
        return {}
    placeholders, params = in_clause(uids)
    rows = db.query(
        f"SELECT * FROM users WHERE uid IN ({placeholders}) AND deleted_at IS NULL",
        **params,
    )
    return {row["uid"]: dict(row) for row in rows}


def count_users():
    if "users" not in db.tables:
        return 0
    return db["users"].count(deleted_at=None)


def create_user(username, email, password_hash, role=None, bio="", created_at=None):
    role = role or (ROLE_ADMIN if count_users() == 0 else ROLE_MEMBER)
    uid = generate_uid()
    row = {
        "uid": uid,
        "username": username,
        "email": (email or "").lower(),
        "password_hash": password_hash,
        "role": role,
        "api_key": generate_uid(),
        "bio": bio,
        "avatar_seed": None,
        "is_active": True,
        "last_seen_at": None,
        "suspended_until": None,
        "suspend_reason": None,
        "terms_version": get_setting("terms_version", "1"),
        "created_at": created_at or now_iso(),
        "updated_at": now_iso(),
        "deleted_at": None,
        "deleted_by": None,
    }
    db["users"].insert(row)
    invalidate_admins_cache()
    return row


def update_user(uid, **fields):
    fields["uid"] = uid
    fields["updated_at"] = now_iso()
    db["users"].update(fields, ["uid"])


def set_role(uid, role):
    update_user(uid, role=role)
    invalidate_admins_cache()


def invalidate_admins_cache():
    bump_cache_version("admins", _admins_cache)


def admin_uids_ordered():
    sync_local_cache("admins", _admins_cache)
    cached = _admins_cache.get("ordered")
    if cached is not None:
        return cached
    if "users" not in db.tables:
        return []
    rows = db["users"].find(role=ROLE_ADMIN, deleted_at=None, order_by=["created_at", "id"])
    ordered = [row["uid"] for row in rows]
    _admins_cache.set("ordered", ordered)
    return ordered


def list_users(limit=200):
    if "users" not in db.tables:
        return []
    return list(db["users"].find(deleted_at=None, order_by=["created_at"], _limit=limit))


def public_people():
    if "users" not in db.tables:
        return []
    return list(db["users"].find(deleted_at=None, is_active=True, order_by=["created_at"]))