103 lines
3.4 KiB
Python
103 lines
3.4 KiB
Python
from typing import Optional, Dict, Any
|
|
from sqlalchemy.orm import Session
|
|
from email_validator import validate_email, EmailNotValidError
|
|
from models.contact_form import ContactForm
|
|
from schemas.contact_form import ContactFormCreate
|
|
|
|
def create_contact_form(db: Session, contact_form_data: ContactFormCreate) -> ContactForm:
|
|
"""
|
|
Creates a new contact form submission in the database after validating inputs.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
contact_form_data (ContactFormCreate): The data for the contact form to create.
|
|
|
|
Returns:
|
|
ContactForm: The newly created contact form object.
|
|
|
|
Raises:
|
|
HTTPException: If any required field is missing or invalid.
|
|
"""
|
|
# Create a new contact form instance
|
|
db_contact_form = ContactForm(
|
|
name=contact_form_data.name,
|
|
email=contact_form_data.email,
|
|
message=contact_form_data.message
|
|
)
|
|
|
|
# Add to database, commit, and refresh
|
|
db.add(db_contact_form)
|
|
db.commit()
|
|
db.refresh(db_contact_form)
|
|
|
|
return db_contact_form
|
|
|
|
def validate_contact_form_data(data: Dict[str, Any]) -> Dict[str, str]:
|
|
"""
|
|
Validates contact form data manually, checking for required fields and valid email format.
|
|
|
|
Args:
|
|
data (Dict[str, Any]): The contact form data to validate.
|
|
|
|
Returns:
|
|
Dict[str, str]: A dictionary of validation errors, empty if validation passes.
|
|
"""
|
|
errors = {}
|
|
|
|
# Check required fields
|
|
if not data.get("name"):
|
|
errors["name"] = "Name is required"
|
|
|
|
if not data.get("email"):
|
|
errors["email"] = "Email is required"
|
|
elif data.get("email"):
|
|
try:
|
|
# Validate email format using email_validator package
|
|
validate_email(data["email"])
|
|
except EmailNotValidError:
|
|
errors["email"] = "Email format is invalid"
|
|
|
|
if not data.get("message"):
|
|
errors["message"] = "Message is required"
|
|
|
|
return errors
|
|
|
|
def get_contact_form_by_id(db: Session, contact_form_id: str) -> Optional[ContactForm]:
|
|
"""
|
|
Retrieves a contact form submission by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
contact_form_id (str): 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 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).offset(skip).limit(limit).all()
|
|
|
|
def get_contact_forms_by_email(db: Session, email: str) -> list[ContactForm]:
|
|
"""
|
|
Retrieves all contact form submissions from a specific email address.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
email (str): The email address to filter by.
|
|
|
|
Returns:
|
|
list[ContactForm]: A list of contact form objects from the specified email.
|
|
"""
|
|
return db.query(ContactForm).filter(ContactForm.email == email).all() |