99 lines
3.1 KiB
Python
99 lines
3.1 KiB
Python
from typing import Optional, Dict, Any
|
|
from sqlalchemy.orm import Session
|
|
from email_validator import validate_email, EmailNotValidError
|
|
from uuid import UUID
|
|
|
|
from models.contactform import ContactForm
|
|
from schemas.contactform import ContactFormCreate, ContactFormSchema
|
|
|
|
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 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 validate_contact_form_data(form_data: Dict[str, Any]) -> Dict[str, str]:
|
|
"""
|
|
Validates the contact form data against required fields and format rules.
|
|
|
|
Args:
|
|
form_data (Dict[str, Any]): The raw contact form data to validate.
|
|
|
|
Returns:
|
|
Dict[str, str]: A dictionary of validation errors, empty if validation passes.
|
|
"""
|
|
errors = {}
|
|
|
|
# Check for required fields
|
|
if not form_data.get("name"):
|
|
errors["name"] = "Name is required"
|
|
|
|
if not form_data.get("email"):
|
|
errors["email"] = "Email is required"
|
|
else:
|
|
# Validate email format
|
|
try:
|
|
validate_email(form_data["email"])
|
|
except EmailNotValidError:
|
|
errors["email"] = "Invalid email format"
|
|
|
|
if not form_data.get("message"):
|
|
errors["message"] = "Message is required"
|
|
|
|
return errors
|
|
|
|
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 object if found, otherwise None.
|
|
"""
|
|
return db.query(ContactForm).filter(ContactForm.id == form_id).first()
|
|
|
|
def format_validation_error_response(errors: Dict[str, str]) -> Dict[str, Any]:
|
|
"""
|
|
Formats validation errors into a standardized response structure.
|
|
|
|
Args:
|
|
errors (Dict[str, str]): Dictionary of field-specific validation errors.
|
|
|
|
Returns:
|
|
Dict[str, Any]: A structured error response.
|
|
"""
|
|
return {
|
|
"status": "error",
|
|
"message": "Validation error",
|
|
"errors": errors
|
|
}
|
|
|
|
def send_notification_email(contact_form: ContactFormSchema) -> bool:
|
|
"""
|
|
Sends a notification email about the new contact form submission.
|
|
This is a placeholder for actual email sending implementation.
|
|
|
|
Args:
|
|
contact_form (ContactFormSchema): The contact form data.
|
|
|
|
Returns:
|
|
bool: True if the email was sent successfully, False otherwise.
|
|
"""
|
|
# This would be implemented with an actual email sending library
|
|
# like smtplib, sendgrid, etc.
|
|
# For now, just return True to simulate successful sending
|
|
return True |