"""Shared helpers for talking to ighub.users.""" from typing import Any from bson import ObjectId from bson.errors import InvalidId from fastapi import HTTPException, status # `password` is stored in plaintext in this database, so it is stripped at the # query layer -- never rely on the response model to hide it (UserOut allows # extra keys so publisher-shaped docs survive the round trip). NO_PASSWORD = {"password": 0} SORTABLE = { "created_on", "last_login", "fname", "lname", "user_id", "email", "role", "status", "department", "total_score", "day_score", "article_count", "queries_count", "available_amount", } def to_object_id(value: str) -> ObjectId | None: try: return ObjectId(value) except (InvalidId, TypeError): return None def ref_filter(user_ref: str) -> dict[str, Any]: """Match a user by _id when the ref is a valid ObjectId, else by user_id.""" oid = to_object_id(user_ref) return {"_id": oid} if oid else {"user_id": user_ref} def not_found(user_ref: str) -> HTTPException: return HTTPException( status_code=status.HTTP_404_NOT_FOUND, detail=f"No user matching '{user_ref}'", )