128 lines
3.6 KiB
Python
128 lines
3.6 KiB
Python
from typing import Optional, Dict, Any
|
|
from sqlalchemy.orm import Session
|
|
from fastapi import HTTPException
|
|
from email_validator import validate_email, EmailNotValidError
|
|
from uuid import UUID
|
|
|
|
from models.contact_form import ContactForm
|
|
from schemas.contact_form import ContactFormCreate
|
|
|
|
def validate_contact_form_data(form_data: Dict[str, Any]) -> Dict[str, str]:
|
|
"""
|
|
Validates contact form data and returns error messages for invalid fields.
|
|
|
|
Args:
|
|
form_data (Dict[str, Any]): The contact form data to validate
|
|
|
|
Returns:
|
|
Dict[str, str]: Dictionary of field errors, empty if validation passes
|
|
"""
|
|
errors = {}
|
|
|
|
# Check for required fields
|
|
if not form_data.get("name"):
|
|
errors["name"] = "Name is required"
|
|
|
|
# Validate email
|
|
if not form_data.get("email"):
|
|
errors["email"] = "Email is required"
|
|
else:
|
|
try:
|
|
validate_email(form_data["email"])
|
|
except EmailNotValidError:
|
|
errors["email"] = "Invalid email format"
|
|
|
|
# Check message
|
|
if not form_data.get("message"):
|
|
errors["message"] = "Message is required"
|
|
|
|
return errors
|
|
|
|
def create_contact_form(db: Session, form_data: ContactFormCreate) -> ContactForm:
|
|
"""
|
|
Creates a new contact form submission in the database.
|
|
|
|
Args:
|
|
db (Session): The database session
|
|
form_data (ContactFormCreate): The validated form data
|
|
|
|
Returns:
|
|
ContactForm: The newly created contact form object
|
|
"""
|
|
db_contact_form = ContactForm(
|
|
name=form_data.name,
|
|
email=form_data.email,
|
|
message=form_data.message
|
|
)
|
|
|
|
db.add(db_contact_form)
|
|
db.commit()
|
|
db.refresh(db_contact_form)
|
|
|
|
return db_contact_form
|
|
|
|
def get_contact_form_by_id(db: Session, form_id: UUID) -> Optional[ContactForm]:
|
|
"""
|
|
Retrieves a contact form submission by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session
|
|
form_id (UUID): The ID of the contact form to retrieve
|
|
|
|
Returns:
|
|
Optional[ContactForm]: The contact form if found, otherwise None
|
|
"""
|
|
return db.query(ContactForm).filter(ContactForm.id == form_id).first()
|
|
|
|
def format_validation_errors(errors: Dict[str, str]) -> Dict[str, Any]:
|
|
"""
|
|
Formats validation errors for the API response.
|
|
|
|
Args:
|
|
errors (Dict[str, str]): Dictionary of field errors
|
|
|
|
Returns:
|
|
Dict[str, Any]: Formatted error response
|
|
"""
|
|
return {
|
|
"status": "error",
|
|
"message": "Validation failed",
|
|
"errors": errors
|
|
}
|
|
|
|
def handle_contact_form_submission(db: Session, form_data: Dict[str, Any]) -> Dict[str, Any]:
|
|
"""
|
|
Validates and processes a contact form submission.
|
|
|
|
Args:
|
|
db (Session): The database session
|
|
form_data (Dict[str, Any]): The raw form data
|
|
|
|
Returns:
|
|
Dict[str, Any]: Response data
|
|
|
|
Raises:
|
|
HTTPException: If validation fails
|
|
"""
|
|
# Validate form data
|
|
validation_errors = validate_contact_form_data(form_data)
|
|
|
|
if validation_errors:
|
|
error_response = format_validation_errors(validation_errors)
|
|
raise HTTPException(status_code=400, detail=error_response)
|
|
|
|
# Create validated form data object
|
|
form_create = ContactFormCreate(
|
|
name=form_data["name"],
|
|
email=form_data["email"],
|
|
message=form_data["message"]
|
|
)
|
|
|
|
# Save to database
|
|
contact_form = create_contact_form(db, form_create)
|
|
|
|
return {
|
|
"status": "success",
|
|
"message": "Contact form submitted successfully",
|
|
"id": str(contact_form.id)
|
|
} |