137 lines
4.2 KiB
Python
137 lines
4.2 KiB
Python
from typing import Dict, Optional, Union, Any
|
|
from fastapi import HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from email_validator import validate_email, EmailNotValidError
|
|
from models.contact_form import ContactForm
|
|
from schemas.contact_form import ContactFormCreate, ContactFormSchema
|
|
|
|
def validate_contact_form_data(
|
|
form_data: Dict[str, Any]
|
|
) -> Dict[str, Optional[str]]:
|
|
"""
|
|
Validates contact form data fields: name, email, and message.
|
|
|
|
Args:
|
|
form_data (Dict[str, Any]): The contact form data to validate
|
|
|
|
Returns:
|
|
Dict[str, Optional[str]]: Dictionary with field names as keys and error messages as values.
|
|
If a field is valid, its error message will be None.
|
|
"""
|
|
errors = {
|
|
"name": None,
|
|
"email": None,
|
|
"message": None
|
|
}
|
|
|
|
# Validate name
|
|
if "name" not in form_data or not form_data["name"]:
|
|
errors["name"] = "Name is required"
|
|
|
|
# Validate email
|
|
if "email" not in form_data or not form_data["email"]:
|
|
errors["email"] = "Email is required"
|
|
else:
|
|
try:
|
|
# Use email_validator to validate email format
|
|
validate_email(form_data["email"])
|
|
except EmailNotValidError:
|
|
errors["email"] = "Invalid email format"
|
|
|
|
# Validate message
|
|
if "message" not in form_data or not form_data["message"]:
|
|
errors["message"] = "Message is required"
|
|
|
|
return errors
|
|
|
|
def validate_and_create_contact_form(
|
|
db: Session,
|
|
form_data: ContactFormCreate
|
|
) -> Union[ContactForm, Dict[str, Optional[str]]]:
|
|
"""
|
|
Validates contact form data and creates a new contact form entry if valid.
|
|
|
|
Args:
|
|
db (Session): The database session
|
|
form_data (ContactFormCreate): The contact form data to validate and create
|
|
|
|
Returns:
|
|
Union[ContactForm, Dict[str, Optional[str]]]:
|
|
- ContactForm object if validation passes
|
|
- Dictionary of validation errors if validation fails
|
|
"""
|
|
# Convert Pydantic model to dict for validation
|
|
data_dict = form_data.dict()
|
|
|
|
# Validate the form data
|
|
validation_errors = validate_contact_form_data(data_dict)
|
|
|
|
# Check if there are any validation errors
|
|
if any(error is not None for error in validation_errors.values()):
|
|
return validation_errors
|
|
|
|
# Create new contact form entry
|
|
db_contact_form = ContactForm(**data_dict)
|
|
db.add(db_contact_form)
|
|
db.commit()
|
|
db.refresh(db_contact_form)
|
|
|
|
return db_contact_form
|
|
|
|
def create_contact_form_entry(
|
|
db: Session,
|
|
form_data: ContactFormCreate
|
|
) -> ContactForm:
|
|
"""
|
|
Creates a new contact form entry in the database.
|
|
|
|
Args:
|
|
db (Session): The database session
|
|
form_data (ContactFormCreate): The validated contact form data
|
|
|
|
Returns:
|
|
ContactForm: The newly created contact form object
|
|
"""
|
|
db_contact_form = ContactForm(**form_data.dict())
|
|
db.add(db_contact_form)
|
|
db.commit()
|
|
db.refresh(db_contact_form)
|
|
return db_contact_form
|
|
|
|
def handle_contact_form_submission(
|
|
db: Session,
|
|
form_data: ContactFormCreate
|
|
) -> ContactFormSchema:
|
|
"""
|
|
Process a contact form submission with validation.
|
|
|
|
Args:
|
|
db (Session): The database session
|
|
form_data (ContactFormCreate): The contact form data to process
|
|
|
|
Returns:
|
|
ContactFormSchema: The created contact form entry
|
|
|
|
Raises:
|
|
HTTPException: If validation fails with specific error messages
|
|
"""
|
|
# Validate the form data
|
|
validation_errors = validate_contact_form_data(form_data.dict())
|
|
|
|
# Build error message if validation fails
|
|
error_messages = []
|
|
for field, error in validation_errors.items():
|
|
if error:
|
|
error_messages.append(f"{field}: {error}")
|
|
|
|
if error_messages:
|
|
raise HTTPException(
|
|
status_code=status.HTTP_400_BAD_REQUEST,
|
|
detail={"errors": error_messages}
|
|
)
|
|
|
|
# Create the contact form entry
|
|
contact_form = create_contact_form_entry(db, form_data)
|
|
|
|
# Return the created contact form
|
|
return ContactFormSchema.from_orm(contact_form) |