80 lines
2.3 KiB
Python
80 lines
2.3 KiB
Python
from typing import Optional
|
|
from pydantic import EmailStr, validator
|
|
from sqlalchemy.orm import Session
|
|
from models.user import User
|
|
from schemas.user import UserCreate
|
|
|
|
def validate_phone_number(phone: str) -> bool:
|
|
"""
|
|
Validate a phone number format.
|
|
|
|
Args:
|
|
phone: The phone number to validate
|
|
|
|
Returns:
|
|
bool: True if phone number format is valid, False otherwise
|
|
"""
|
|
# Implement phone number validation logic here
|
|
# Example: Check if the phone number starts with a '+' and contains only digits
|
|
pattern = r'^\+?\d+$'
|
|
return bool(re.match(pattern, phone))
|
|
|
|
def get_user_by_phone(db: Session, phone: str) -> Optional[User]:
|
|
"""
|
|
Get a user by their phone number.
|
|
|
|
Args:
|
|
db: Database session
|
|
phone: Phone number to search for
|
|
|
|
Returns:
|
|
User object if found, None otherwise
|
|
"""
|
|
return db.query(User).filter(User.phone == phone).first()
|
|
|
|
def create_user(db: Session, user_data: UserCreate) -> User:
|
|
"""
|
|
Create a new user with phone number and validation.
|
|
|
|
Args:
|
|
db: Database session
|
|
user_data: User data for creation
|
|
|
|
Returns:
|
|
User object if created successfully
|
|
|
|
Raises:
|
|
ValueError: If email or phone number is invalid
|
|
Exception: If user with the same email or phone number already exists
|
|
"""
|
|
# Check if email is valid
|
|
if not validate_email(user_data.email):
|
|
raise ValueError("Invalid email format")
|
|
|
|
# Check if phone number is valid
|
|
if not validate_phone_number(user_data.phone):
|
|
raise ValueError("Invalid phone number format")
|
|
|
|
# Check if user with the same email already exists
|
|
existing_user = get_user_by_email(db, user_data.email)
|
|
if existing_user:
|
|
raise Exception("Email already registered")
|
|
|
|
# Check if user with the same phone number already exists
|
|
existing_user = get_user_by_phone(db, user_data.phone)
|
|
if existing_user:
|
|
raise Exception("Phone number already registered")
|
|
|
|
# Create the user
|
|
db_user = User(
|
|
username=user_data.username,
|
|
email=user_data.email,
|
|
phone=user_data.phone,
|
|
password=user_data.password # In a real app, hash this password
|
|
)
|
|
|
|
db.add(db_user)
|
|
db.commit()
|
|
db.refresh(db_user)
|
|
|
|
return db_user |