125 lines
3.2 KiB
Python
125 lines
3.2 KiB
Python
from typing import List, Optional, Dict, Union
|
|
from datetime import datetime
|
|
import re
|
|
from sqlalchemy.orm import Session
|
|
from pydantic import BaseModel, EmailStr
|
|
from models.person import Person
|
|
|
|
def validate_phone_number(phone: str) -> bool:
|
|
"""
|
|
Validate phone number format.
|
|
|
|
Args:
|
|
phone: Phone number string to validate
|
|
|
|
Returns:
|
|
bool: True if valid format, False otherwise
|
|
"""
|
|
pattern = r'^\+?1?\d{9,15}$'
|
|
return bool(re.match(pattern, phone))
|
|
|
|
def get_people_over_50(db: Session) -> List[Person]:
|
|
"""
|
|
Get all people over 50 years old.
|
|
|
|
Args:
|
|
db: Database session
|
|
|
|
Returns:
|
|
List of Person objects over 50
|
|
"""
|
|
return db.query(Person).filter(Person.age > 50, Person.is_active == True).all()
|
|
|
|
def format_person_name(first_name: str, last_name: str) -> str:
|
|
"""
|
|
Format person's full name.
|
|
|
|
Args:
|
|
first_name: Person's first name
|
|
last_name: Person's last name
|
|
|
|
Returns:
|
|
Formatted full name string
|
|
"""
|
|
return f"{first_name.strip().title()} {last_name.strip().title()}"
|
|
|
|
def create_person_safely(
|
|
db: Session,
|
|
first_name: str,
|
|
last_name: str,
|
|
age: int,
|
|
email: str,
|
|
phone: Optional[str] = None
|
|
) -> Union[Person, Dict[str, str]]:
|
|
"""
|
|
Create new person with validation.
|
|
|
|
Args:
|
|
db: Database session
|
|
first_name: Person's first name
|
|
last_name: Person's last name
|
|
age: Person's age
|
|
email: Person's email
|
|
phone: Optional phone number
|
|
|
|
Returns:
|
|
Person object if created successfully, error dict otherwise
|
|
"""
|
|
# Validate required fields
|
|
if not all([first_name, last_name, age, email]):
|
|
return {"error": "Missing required fields"}
|
|
|
|
# Validate age
|
|
if not isinstance(age, int) or age < 0:
|
|
return {"error": "Invalid age"}
|
|
|
|
# Validate email format
|
|
email_pattern = r'^[a-zA-Z0-9_.+-]+@[a-zA-Z0-9-]+\.[a-zA-Z0-9-.]+$'
|
|
if not re.match(email_pattern, email):
|
|
return {"error": "Invalid email format"}
|
|
|
|
# Check if email exists
|
|
if db.query(Person).filter(Person.email == email).first():
|
|
return {"error": "Email already exists"}
|
|
|
|
# Validate phone if provided
|
|
if phone and not validate_phone_number(phone):
|
|
return {"error": "Invalid phone number format"}
|
|
|
|
person = Person(
|
|
first_name=first_name,
|
|
last_name=last_name,
|
|
age=age,
|
|
email=email,
|
|
phone=phone,
|
|
is_active=True
|
|
)
|
|
|
|
db.add(person)
|
|
db.commit()
|
|
db.refresh(person)
|
|
|
|
return person
|
|
|
|
def get_active_seniors(db: Session, min_age: int = 50) -> List[Dict]:
|
|
"""
|
|
Get formatted list of active people over minimum age.
|
|
|
|
Args:
|
|
db: Database session
|
|
min_age: Minimum age to filter by
|
|
|
|
Returns:
|
|
List of formatted person dictionaries
|
|
"""
|
|
seniors = db.query(Person).filter(
|
|
Person.age >= min_age,
|
|
Person.is_active == True
|
|
).all()
|
|
|
|
return [{
|
|
"full_name": format_person_name(p.first_name, p.last_name),
|
|
"age": p.age,
|
|
"email": p.email,
|
|
"phone": p.phone
|
|
} for p in seniors] |