class user_util(): def get_user_id(token): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') credentials_exception = HTTPException( status_code=status.HTTP_401_UNAUTHORIZED, detail="Could not validate credentials", headers={"WWW-Authenticate": "Bearer"}, ) try: payload = jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM]) username: str = payload.get("sub") if username is None: raise credentials_exception token_data = models.TokenData(username=username) except JWTError: raise credentials_exception user = get_user(username=token_data.username) if user is None: raise credentials_exception user_id = first(db.query('SELECT * FROM users where username="' + user.username+'"'))['id'] return user_id def check_user_exists(username): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') if int(next(iter(db.query('SELECT COUNT(*) FROM AI_anchor.users WHERE username = "'+username+'"')))['COUNT(*)']) > 0: return True else: return False def get_user(username: str): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') if not check_user_exists(username): # if user don't exist return False user_dict = next( iter(db.query('SELECT * FROM AI_anchor.users where username ="'+username+'"'))) user = models.User(**user_dict) return user def user_register(user): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') table = db['users'] user.password = get_password_hash(user.password) table.insert(dict(user)) def get_password_hash(password): return pwd_context.hash(password) def verify_password(plain_password, hashed_password): return pwd_context.verify(plain_password, hashed_password) def authenticate_user(username: str, password: str): db = dataset.connect('mysql://choozmo:pAssw0rd@db.ptt.cx:3306/AI_anchor?charset=utf8mb4') if not check_user_exists(username): # if user don't exist return False user_dict = next(iter(db.query('SELECT * FROM AI_anchor.users where username ="'+username+'"'))) user = models.User(**user_dict) if not verify_password(password, user.password): return False return user