133 lines
4.2 KiB
Python
133 lines
4.2 KiB
Python
from typing import Dict, Any
|
|
from sqlalchemy.orm import Session
|
|
from models.contact import Contact
|
|
from schemas.contact import ContactCreate, ContactSchema
|
|
from fastapi import HTTPException, status
|
|
from pydantic import EmailStr, ValidationError
|
|
|
|
def validate_contact_data(contact_data: Dict[str, Any]) -> Dict[str, str]:
|
|
"""
|
|
Validates contact form submission data with enhanced validation.
|
|
|
|
Args:
|
|
contact_data (Dict[str, Any]): The contact form data to validate.
|
|
|
|
Returns:
|
|
Dict[str, str]: Dictionary of validation errors, empty if valid.
|
|
"""
|
|
errors = {}
|
|
|
|
# Validate name
|
|
if not contact_data.get("name"):
|
|
errors["name"] = "Name is required"
|
|
elif not contact_data["name"].strip():
|
|
errors["name"] = "Name cannot consist of only whitespace"
|
|
|
|
# Validate email with enhanced validation
|
|
if not contact_data.get("email"):
|
|
errors["email"] = "Email is required"
|
|
else:
|
|
try:
|
|
EmailStr.validate(contact_data["email"])
|
|
except ValidationError:
|
|
errors["email"] = "Invalid email format - please provide a valid email address"
|
|
|
|
# Validate message
|
|
if not contact_data.get("message"):
|
|
errors["message"] = "Message is required"
|
|
elif not contact_data["message"].strip():
|
|
errors["message"] = "Message cannot consist of only whitespace"
|
|
|
|
return errors
|
|
|
|
def create_contact(db: Session, contact_data: ContactCreate) -> Contact:
|
|
"""
|
|
Creates a new contact submission in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
contact_data (ContactCreate): The validated contact form data.
|
|
|
|
Returns:
|
|
Contact: The newly created contact object.
|
|
|
|
Raises:
|
|
HTTPException: If there are validation errors with specific field details.
|
|
"""
|
|
try:
|
|
# ContactCreate schema will handle validation
|
|
validated_data = contact_data.dict()
|
|
db_contact = Contact(**validated_data)
|
|
db.add(db_contact)
|
|
db.commit()
|
|
db.refresh(db_contact)
|
|
return db_contact
|
|
except ValidationError as e:
|
|
errors = {error["loc"][0]: error["msg"] for error in e.errors()}
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=errors
|
|
)
|
|
|
|
def format_contact_response(contact: Contact) -> ContactSchema:
|
|
"""
|
|
Formats a contact database object into the response schema.
|
|
|
|
Args:
|
|
contact (Contact): The contact database object.
|
|
|
|
Returns:
|
|
ContactSchema: The formatted contact response.
|
|
"""
|
|
return ContactSchema.from_orm(contact)
|
|
|
|
def sanitize_contact_input(contact_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Sanitizes contact form input data with enhanced validation for required fields.
|
|
|
|
Args:
|
|
contact_data (Dict[str, Any]): Raw contact form data.
|
|
|
|
Returns:
|
|
Dict[str, Any]: Sanitized contact form data.
|
|
|
|
Raises:
|
|
HTTPException: If required fields are missing or invalid with specific field details.
|
|
"""
|
|
errors = {}
|
|
sanitized = {}
|
|
|
|
# Sanitize and validate name
|
|
if "name" in contact_data:
|
|
sanitized["name"] = contact_data["name"].strip()
|
|
if not sanitized["name"]:
|
|
errors["name"] = "Name cannot be empty or consist of only whitespace"
|
|
else:
|
|
errors["name"] = "Name is required"
|
|
|
|
# Sanitize and validate email
|
|
if "email" in contact_data:
|
|
email = contact_data["email"].lower().strip()
|
|
try:
|
|
EmailStr.validate(email)
|
|
sanitized["email"] = email
|
|
except ValidationError:
|
|
errors["email"] = "Invalid email format"
|
|
else:
|
|
errors["email"] = "Email is required"
|
|
|
|
# Sanitize and validate message
|
|
if "message" in contact_data:
|
|
sanitized["message"] = contact_data["message"].strip()
|
|
if not sanitized["message"]:
|
|
errors["message"] = "Message cannot be empty or consist of only whitespace"
|
|
else:
|
|
errors["message"] = "Message is required"
|
|
|
|
if errors:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail=errors
|
|
)
|
|
|
|
return sanitized |