from datetime import datetime, timedelta import uuid from pathlib import Path from sqlalchemy.orm import joinedload from sqlalchemy.future import select from core.config import BASE_PATH, get_db, settings from core.email.config import SendForgotPasswordEmail from core.security.authentication import verify_password, create_access_token, hash_password from core.security.exceptions import CoreDBError, GenericError from core.utility import send_email from models import User, UserProfile, UserWallet, Participant, Parent, PasswordResetToken from models.user import UserTypesEnum from schema.user import ProfileResponseSchema, ParentSchema from models import Staff class UserService: def create_user(self, request): with get_db() as db: payload = request.model_dump(exclude=['password', 'confirm_password','last_name','dob']) payload['hashed_password'] = hash_password(request.confirm_password) payload['sur_name'] = request.last_name try: new_user = User(**payload) db.add(new_user) db.commit() db.refresh(new_user) return new_user except Exception as e: db.rollback() raise CoreDBError(f"Could not create account: {e}") def validate_user_for_login(self, request): with get_db() as db: models={ 'admin':User, 'parent':Parent, 'participant':Participant, 'guest':User, 'staff':Staff, } model=models[request.user_type] user = db.query(model).where(model.email == request.username, model.deleted_at.is_(None), model.is_active == True).first() if user is None: raise GenericError(status_code=404, exc="User not exists!") if not verify_password(request.password, user.hashed_password): raise GenericError(status_code=404, exc="User not exists!") payload = {'id': user.id, 'email': user.email, 'first_name': user.first_name, 'user_type': request.user_type,"mobile_number":user.mobile_number} if request.user_type =='staff': payload['sur_name']=user.last_name else: payload["sur_name"]=user.sur_name payload['token'] = create_access_token(payload) if request.user_type in ['participant','parent']: payload['age']=user.age payload['is_above_18']=user.age>18 else: payload['is_above_18']=False return payload def update_profile(self, request, user): with get_db() as db: payload = request.model_dump() payload['user_id'] = user['id'] user_profile = db.query(UserProfile).filter(UserProfile.user_id == user['id']).first() if not user_profile: # create new_profile = UserProfile(**payload) db.add(new_profile) db.commit() db.refresh(new_profile) else: result = (db.query(UserProfile).filter(UserProfile.user_id == user['id'], UserProfile.deleted_at.is_(None)).update(payload, synchronize_session="fetch")) db.commit() if result == 0: raise GenericError(status_code=422, exc="Could not update profile") def fetch_profile(self, user): with get_db() as db: user_profile = db.query(UserProfile).filter(UserProfile.user_id == user['id']).first() if user_profile: return ProfileResponseSchema.model_validate(user_profile).model_dump() else: return None async def upload_profile_image(self, request, user): file_extension = Path(request.image.filename).suffix # Extract the file extension new_filename = f"{uuid.uuid4()}{file_extension}" file_location = BASE_PATH / f'uploads/{new_filename}' content = await request.image.read() with open(file_location, "wb") as f: f.write(content) models = { 'admin': User, 'parent': Parent, 'participant': Participant, 'guest': User, 'staff': Staff, } model = models[request.user_type] print(model) with get_db() as db: result = db.query(model).filter(model.id == request.user_id).update( {'profile_image': new_filename}, synchronize_session="fetch") db.commit() if result == 0: raise GenericError(status_code=500, exc="Could not upload profile image") def create_wallet(self, db, user): wallet = UserWallet(user_id=user.id) db.add(wallet) db.commit() def get_all_parent_user(self): with get_db() as db: users = db.query(User).options(joinedload(User.user_profile)).filter(User.user_type == 'parent.py').all() return [ParentSchema.model_validate(user) for user in users] def send_password_reset_email(self, request): with get_db() as db: user_types = UserTypesEnum(value=request.user_type) UserTypeModel = user_types.model result = db.execute(select(UserTypeModel).where(UserTypeModel.email == request.email)) user = result.scalar_one_or_none() if not user: raise GenericError(status_code=404, exc="User not found") db.query(PasswordResetToken).filter( PasswordResetToken.user_id == user.id ).delete(synchronize_session="fetch") token = str(uuid.uuid4()) expiry = datetime.utcnow() + timedelta(minutes=settings.reset_password_token_time_out) reset_entry = PasswordResetToken( token=token, expiry=expiry, user_id=user.id, user_type=user_types.name ) db.add(reset_entry) db.commit() email = SendForgotPasswordEmail(token_instance=reset_entry) status = email.send_email() return status def reset_password(self, request): with get_db() as db: reset_entry = db.execute( select( PasswordResetToken ).where( PasswordResetToken.token == request.token ) ) UserTypeModel = UserTypesEnum(value=request.user_type).model reset_token = reset_entry.scalar_one_or_none() if not reset_token or reset_token.is_expired(): raise GenericError( status_code=400, exc="Invalid or expired token" ) hashed_password = hash_password(request.new_password) db.query(UserTypeModel).filter( UserTypeModel.id == reset_token.user_id, ).update( {'hashed_password': hashed_password}, synchronize_session="fetch" ) db.query(PasswordResetToken).filter( PasswordResetToken.user_id == reset_token.user_id ).delete(synchronize_session="fetch") db.commit() return True async def change_password(self, user, data): UserTypeModel = UserTypesEnum(value=data.user_type).model with get_db() as db: result = db.execute(select(UserTypeModel).where(UserTypeModel.id == user.get('id'))) user = result.scalar_one_or_none() if not user: raise GenericError(status_code=400, exc='User not found') if not verify_password(data.old_password, user.hashed_password): raise GenericError( status_code=401, exc="Old password is incorrect" ) hashed_password = hash_password(data.new_password) user.hashed_password = hashed_password db.add(user) db.commit() def toggle_user_active(self, user_id, user_type): with get_db() as db: if user_type.lower() == 'parent': item = db.query(Parent).filter(Parent.id == user_id).first() elif user_type.lower() == 'participant': item = db.query(Participant).filter(Participant.id == user_id).first() else: raise GenericError(status_code=400, exc="Unsupported user type") if not item: raise GenericError(status_code=404, exc="Item not found") if hasattr(item, 'is_active'): item.is_active = not item.is_active db.commit() return item.is_active else: raise GenericError(status_code=400, exc="Could not disable user ")