diff --git a/helpers/user_helpers.py b/helpers/user_helpers.py index 7893023..4431be6 100644 --- a/helpers/user_helpers.py +++ b/helpers/user_helpers.py @@ -2,133 +2,137 @@ from typing import List, Dict, Optional, Union, Any from datetime import datetime import re from sqlalchemy.orm import Session +from sqlalchemy import or_ from models.user import User from schemas.user import UserCreate, UserUpdate +from passlib.context import CryptContext -def validate_phone_number(phone: Optional[str]) -> bool: +pwd_context = CryptContext(schemes=["bcrypt"], deprecated="auto") + +def validate_username(username: str) -> bool: """ - Validate phone number format if provided. + Validate username format and length. Args: - phone: Phone number to validate + username: Username to validate Returns: - bool: True if format is valid or phone is None, False otherwise + bool: True if valid, False otherwise """ - if not phone: - return True - pattern = r'^\+?1?\d{9,15}$' - return bool(re.match(pattern, phone)) + pattern = r'^[a-zA-Z0-9_]{3,32}$' + return bool(re.match(pattern, username)) -def validate_email(email: str) -> bool: +def hash_password(password: str) -> str: """ - Validate email address format. + Hash a password using bcrypt. Args: - email: Email to validate + password: Plain text password Returns: - bool: True if format is valid, False otherwise + str: Hashed password """ - pattern = r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$' - return bool(re.match(pattern, email)) + return pwd_context.hash(password) -def format_user_data(user: User) -> Dict[str, Any]: +def verify_password(plain_password: str, hashed_password: str) -> bool: """ - Format user data for API response. + Verify a password against its hash. + + Args: + plain_password: Password to verify + hashed_password: Stored hash to check against + + Returns: + bool: True if password matches + """ + return pwd_context.verify(plain_password, hashed_password) + +def search_users( + db: Session, + search_term: str, + skip: int = 0, + limit: int = 10 +) -> List[User]: + """ + Search for users by username, first name, or last name. + + Args: + db: Database session + search_term: Term to search for + skip: Number of records to skip + limit: Max number of records to return + + Returns: + List of matching users + """ + return db.query(User).filter( + or_( + User.username.ilike(f"%{search_term}%"), + User.first_name.ilike(f"%{search_term}%"), + User.last_name.ilike(f"%{search_term}%") + ) + ).offset(skip).limit(limit).all() + +def update_last_login(db: Session, user: User) -> User: + """ + Update user's last login timestamp. + + Args: + db: Database session + user: User to update + + Returns: + Updated user object + """ + user.last_login = datetime.utcnow() + db.commit() + db.refresh(user) + return user + +def format_user_profile(user: User) -> Dict[str, Any]: + """ + Format user data for profile display. Args: user: User object to format Returns: - Dict containing formatted user data + Dict containing formatted user profile data """ return { - "name": user.name, - "address": user.address, - "phone": user.phone if user.phone else None, - "email": user.email, - "is_active": user.is_active, - "created_at": user.created_at.isoformat() if hasattr(user, 'created_at') else None + "username": user.username, + "full_name": f"{user.first_name or ''} {user.last_name or ''}".strip(), + "bio": user.bio or "No bio available", + "profile_picture": user.profile_picture or "default.jpg", + "relationship_status": user.relationship_status or "Not specified", + "interests": user.interests.split(",") if user.interests else [], + "last_active": user.last_login.isoformat() if user.last_login else None } -def create_user_safely(db: Session, user_data: UserCreate) -> Union[User, Dict[str, str]]: +def validate_user_update(user_data: UserUpdate) -> Dict[str, str]: """ - Create new user with validation and error handling. + Validate user update data. Args: - db: Database session - user_data: User creation data + user_data: Update data to validate Returns: - User object if created successfully, error dict otherwise + Dict with error messages if validation fails """ - if not validate_email(user_data.email): - return {"error": "Invalid email format"} + errors = {} + + if user_data.username and not validate_username(user_data.username): + errors["username"] = "Invalid username format" - if not validate_phone_number(user_data.phone): - return {"error": "Invalid phone number format"} - - existing_user = db.query(User).filter(User.email == user_data.email).first() - if existing_user: - return {"error": "Email already registered"} - - db_user = User( - name=user_data.name, - address=user_data.address, - phone=user_data.phone, - email=user_data.email, - is_active=True - ) - - db.add(db_user) - db.commit() - db.refresh(db_user) - - return db_user - -def update_user_safely( - db: Session, - user: User, - user_data: UserUpdate -) -> Union[User, Dict[str, str]]: - """ - Update user data with validation. - - Args: - db: Database session - user: Existing user to update - user_data: Update data + if user_data.email and not validate_email(user_data.email): + errors["email"] = "Invalid email format" - Returns: - Updated user object if successful, error dict otherwise - """ - if user_data.email and user_data.email != user.email: - if not validate_email(user_data.email): - return {"error": "Invalid email format"} - if db.query(User).filter(User.email == user_data.email).first(): - return {"error": "Email already exists"} - - if user_data.phone and not validate_phone_number(user_data.phone): - return {"error": "Invalid phone number format"} - - for field, value in user_data.dict(exclude_unset=True).items(): - setattr(user, field, value) - - db.commit() - db.refresh(user) - return user - -def get_active_users(db: Session, skip: int = 0, limit: int = 100) -> List[User]: - """ - Get paginated list of active users. - - Args: - db: Database session - skip: Number of records to skip - limit: Max number of records to return + if user_data.profile_picture and not user_data.profile_picture.endswith(('.jpg','.png','.jpeg')): + errors["profile_picture"] = "Invalid image format" - Returns: - List of active user objects - """ - return db.query(User).filter(User.is_active == True).offset(skip).limit(limit).all() \ No newline at end of file + if user_data.relationship_status and user_data.relationship_status not in [ + "Single", "In a relationship", "Married", "It's complicated" + ]: + errors["relationship_status"] = "Invalid relationship status" + + return errors \ No newline at end of file