Spaces:
Sleeping
Sleeping
| from fastapi import FastAPI,HTTPException, Header | |
| from gamification.objects import * | |
| from gamification.logic import * | |
| from jwtcoding import decode_jwt | |
| from tokenManagement import verify_access_token | |
| class EditableUserLevel(BaseModel): | |
| levelId : str | |
| maxPoints : Optional[int]= None | |
| minPoints : Optional[int]= None | |
| levelName : Optional[str]= None | |
| careerPath : Optional[str]= None | |
| levelNumber : Optional[int]= None | |
| # To convert MongoDB ObjectId to string, if needed | |
| class Config: | |
| json_encoders = { | |
| ObjectId: str | |
| } | |
| gamification = FastAPI() | |
| def create_feedback(feedback:Feedback,authorization: str = Header(...))->bool: | |
| token = authorization.split("Bearer ")[-1] | |
| # Here, you would validate the token (e.g., check with a JWT library) | |
| decoded_user_id,decoded_access_token = decode_jwt(token) | |
| is_valid = verify_access_token(db_uri=MONGO_URI, user_id=decoded_user_id, access_token=decoded_access_token) | |
| if is_valid != True: # Example check | |
| raise HTTPException(status_code=401, detail="Invalid token") | |
| try: | |
| user_feedback = UserFeedback(userId=decoded_user_id,respondedAt=datetime.now(), feedback=feedback.feedback,rating=feedback.rating) | |
| result = create_feedback_func(user_feedback) | |
| return result | |
| except Exception as e: | |
| raise HTTPException(status_code=500,detail=f"{e}") | |
| def get_leaderboard_details(): | |
| pass | |
| def get_points(authorization: str = Header(...))->IndividualUserLevel: | |
| # Extract the token from the Authorization header (Bearer token) | |
| token = authorization.split("Bearer ")[-1] | |
| # Here, you would validate the token (e.g., check with a JWT library) | |
| decoded_user_id,decoded_access_token = decode_jwt(token) | |
| is_valid = verify_access_token(db_uri=MONGO_URI, user_id=decoded_user_id, access_token=decoded_access_token) | |
| if is_valid != True: # Example check | |
| raise HTTPException(status_code=401, detail="Invalid token") | |
| # do thing u want here | |
| points = get_all_points_func(userId=decoded_user_id) | |
| return points.model_dump(exclude_none=True) | |
| def get_points(authorization: str = Header(...))->SimpleIndividualUserLevel: | |
| # Extract the token from the Authorization header (Bearer token) | |
| token = authorization.split("Bearer ")[-1] | |
| # Here, you would validate the token (e.g., check with a JWT library) | |
| decoded_user_id,decoded_access_token = decode_jwt(token) | |
| is_valid = verify_access_token(db_uri=MONGO_URI, user_id=decoded_user_id, access_token=decoded_access_token) | |
| if is_valid != True: # Example check | |
| raise HTTPException(status_code=401, detail="Invalid token") | |
| # do thing u want here | |
| points = get_all_simple_points_func(userId=decoded_user_id) | |
| return points.model_dump(exclude_none=True) | |
| def get_customer_information(): | |
| try: | |
| result = get_all_customer_info() | |
| return result | |
| except Exception as e: | |
| raise HTTPException(status_code=500,detail=f"{e}") | |
| def get_feedback()->List[UserFeedback]: | |
| feedbacks = get_all_feedback_func() | |
| return feedbacks | |
| def create_level(level_obj:UserLevel)->bool: | |
| result = create_level_func(document=level_obj) | |
| return result | |
| def get_level_details_and_information()->List[UserLevel]: | |
| levels= get_all_levels_func() | |
| return levels | |
| def edit_level(level_obj:EditableUserLevel)->int: | |
| result = edit_level_func(level_id=level_obj.levelId,levelName=level_obj.levelName,careerPath=level_obj.careerPath,minPoints=level_obj.minPoints,maxPoints=level_obj.maxPoints,levelNumber=level_obj.levelNumber) | |
| return {"message":result} | |
| def delete_level(levelId)->bool: | |
| try: | |
| result = delete_level_func(level_id=levelId) | |
| return {"message":result} | |
| except Exception as e: | |
| raise HTTPException(status_code=500,detail=f"{e}") | |
| def get_leaderboard()->List[Ranker]: | |
| try: | |
| list_of_rankers = [] | |
| result = get_top_30() | |
| list_of_rankers = [Ranker(**ranker) for ranker in result] | |
| return list_of_rankers | |
| except Exception as e: | |
| raise HTTPException(status_code=500,detail=f"{e}") |