97 lines
2.5 KiB
Python
97 lines
2.5 KiB
Python
import uuid
|
|
import postgres
|
|
import bcrypt
|
|
|
|
|
|
def getUserUUID(username):
|
|
userRecord = postgres.select_one("users", {"username": username})
|
|
if userRecord:
|
|
return userRecord["id"]
|
|
return False
|
|
|
|
|
|
def getUserFirstName(userUUID):
|
|
userRecord = postgres.select_one("users", {"id": userUUID})
|
|
if userRecord:
|
|
return userRecord.get("username")
|
|
return None
|
|
|
|
|
|
def isUsernameAvailable(username):
|
|
return not postgres.exists("users", {"username": username})
|
|
|
|
|
|
def doesUserUUIDExist(userUUID):
|
|
return postgres.exists("users", {"id": userUUID})
|
|
|
|
|
|
def registerUser(username, password, data=None):
|
|
if isUsernameAvailable(username):
|
|
hashed_pass = bcrypt.hashpw(password.encode("utf-8"), bcrypt.gensalt())
|
|
user_data = {
|
|
"id": str(uuid.uuid4()),
|
|
"username": username,
|
|
"password_hashed": hashed_pass,
|
|
}
|
|
if data:
|
|
user_data.update(data)
|
|
createUser(user_data)
|
|
return True
|
|
return False
|
|
|
|
|
|
def updateUser(userUUID, data_dict):
|
|
user = postgres.select_one("users", {"id": userUUID})
|
|
if not user:
|
|
return False
|
|
blocked = {"id", "password_hashed", "created_at"}
|
|
allowed = set(user.keys()) - blocked
|
|
updates = {k: v for k, v in data_dict.items() if k in allowed}
|
|
if not updates:
|
|
return False
|
|
postgres.update("users", updates, {"id": userUUID})
|
|
return True
|
|
|
|
|
|
def changePassword(userUUID, new_password):
|
|
user = postgres.select_one("users", {"id": userUUID})
|
|
if not user:
|
|
return False
|
|
hashed = bcrypt.hashpw(new_password.encode("utf-8"), bcrypt.gensalt())
|
|
postgres.update("users", {"password_hashed": hashed}, {"id": userUUID})
|
|
return True
|
|
|
|
|
|
def deleteUser(userUUID):
|
|
user = postgres.select_one("users", {"id": userUUID})
|
|
if not user:
|
|
return False
|
|
postgres.delete("users", {"id": userUUID})
|
|
return True
|
|
|
|
|
|
def createUser(data_dict):
|
|
user_schema = {
|
|
"id": None,
|
|
"username": None,
|
|
"password_hashed": None,
|
|
"created_at": None,
|
|
}
|
|
for key in user_schema:
|
|
if key in data_dict:
|
|
user_schema[key] = data_dict[key]
|
|
|
|
is_valid, errors = validateUser(user_schema)
|
|
if not is_valid:
|
|
raise ValueError(f"Invalid user data: {', '.join(errors)}")
|
|
|
|
postgres.insert("users", user_schema)
|
|
|
|
|
|
def validateUser(user):
|
|
required = ["id", "username", "password_hashed"]
|
|
missing = [f for f in required if f not in user or user[f] is None]
|
|
if missing:
|
|
return False, missing
|
|
return True, []
|