108 lines
3.6 KiB
Python
108 lines
3.6 KiB
Python
from typing import Optional, List, Dict
|
|
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, contact_form: ContactFormCreate) -> ContactForm:
|
|
"""
|
|
Creates a new contact form submission in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
contact_form (ContactFormCreate): The contact form data to create.
|
|
|
|
Returns:
|
|
ContactForm: The newly created contact form object.
|
|
"""
|
|
db_contact_form = ContactForm(**contact_form.dict())
|
|
db.add(db_contact_form)
|
|
db.commit()
|
|
db.refresh(db_contact_form)
|
|
return db_contact_form
|
|
|
|
def validate_contact_form_data(name: str, email: str, message: str) -> Dict[str, str]:
|
|
"""
|
|
Validates contact form data fields.
|
|
|
|
Args:
|
|
name (str): The name field to validate.
|
|
email (str): The email field to validate.
|
|
message (str): The message field to validate.
|
|
|
|
Returns:
|
|
Dict[str, str]: Dictionary containing error messages if validation fails, empty if valid.
|
|
"""
|
|
errors = {}
|
|
|
|
# Validate name
|
|
if not name or name.strip() == "":
|
|
errors["name"] = "Name is required"
|
|
|
|
# Validate email
|
|
if not email:
|
|
errors["email"] = "Email is required"
|
|
else:
|
|
try:
|
|
# Use email_validator to validate email format
|
|
validate_email(email)
|
|
except EmailNotValidError:
|
|
errors["email"] = "Email is not valid"
|
|
|
|
# Validate message
|
|
if not message or message.strip() == "":
|
|
errors["message"] = "Message is required"
|
|
|
|
return errors
|
|
|
|
def get_all_contact_forms(db: Session, skip: int = 0, limit: int = 100) -> List[ContactForm]:
|
|
"""
|
|
Retrieves all contact form submissions with pagination.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
skip (int, optional): Number of records to skip. Defaults to 0.
|
|
limit (int, optional): Maximum number of records to return. Defaults to 100.
|
|
|
|
Returns:
|
|
List[ContactForm]: A list of contact form objects.
|
|
"""
|
|
return db.query(ContactForm).order_by(ContactForm.created_at.desc()).offset(skip).limit(limit).all()
|
|
|
|
def get_contact_form_by_id(db: Session, contact_form_id: UUID) -> Optional[ContactForm]:
|
|
"""
|
|
Retrieves a single contact form submission by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
contact_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 == contact_form_id).first()
|
|
|
|
def send_notification_email(contact_form: ContactFormSchema) -> bool:
|
|
"""
|
|
Sends a notification email when a new contact form is submitted.
|
|
This is a placeholder function - implementation would depend on the email service being used.
|
|
|
|
Args:
|
|
contact_form (ContactFormSchema): The contact form data.
|
|
|
|
Returns:
|
|
bool: True if email was sent successfully, False otherwise.
|
|
"""
|
|
# This would be implemented with your email service of choice
|
|
# For example, using SMTP, SendGrid, Mailgun, etc.
|
|
|
|
# Placeholder implementation
|
|
try:
|
|
# Code to send email would go here
|
|
print(f"Notification email about contact from {contact_form.name} <{contact_form.email}> would be sent here")
|
|
return True
|
|
except Exception:
|
|
# Log the error
|
|
return False |