90 lines
3.3 KiB
Python
90 lines
3.3 KiB
Python
from typing import List, Optional
|
|
from sqlalchemy.orm import Session
|
|
import re
|
|
from models.contact_form import ContactForm
|
|
from schemas.contact_form import ContactFormCreate, ContactFormUpdate, ContactFormSchema
|
|
|
|
def get_contact_forms(db: Session) -> List[ContactFormSchema]:
|
|
"""
|
|
Retrieves all contact forms from the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
|
|
Returns:
|
|
List[ContactFormSchema]: A list of all contact form objects.
|
|
"""
|
|
return db.query(ContactForm).all()
|
|
|
|
def get_contact_form_by_id(db: Session, contact_form_id: str) -> Optional[ContactFormSchema]:
|
|
"""
|
|
Retrieves a single contact form by its ID.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
contact_form_id (str): The ID of the contact form to retrieve.
|
|
|
|
Returns:
|
|
Optional[ContactFormSchema]: The contact form object if found, otherwise None.
|
|
"""
|
|
return db.query(ContactForm).filter(ContactForm.id == contact_form_id).first()
|
|
|
|
def create_contact_form(db: Session, contact_form_data: ContactFormCreate) -> ContactFormSchema:
|
|
"""
|
|
Creates a new contact form in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
contact_form_data (ContactFormCreate): The data for the contact form to create.
|
|
|
|
Returns:
|
|
ContactFormSchema: The newly created contact form object.
|
|
"""
|
|
validate_contact_form_data(contact_form_data)
|
|
db_contact_form = ContactForm(**contact_form_data.dict())
|
|
db.add(db_contact_form)
|
|
db.commit()
|
|
db.refresh(db_contact_form)
|
|
return db_contact_form
|
|
|
|
def update_contact_form(db: Session, contact_form_id: str, contact_form_data: ContactFormUpdate) -> Optional[ContactFormSchema]:
|
|
"""
|
|
Updates an existing contact form in the database.
|
|
|
|
Args:
|
|
db (Session): The database session.
|
|
contact_form_id (str): The ID of the contact form to update.
|
|
contact_form_data (ContactFormUpdate): The updated data for the contact form.
|
|
|
|
Returns:
|
|
Optional[ContactFormSchema]: The updated contact form object if found, otherwise None.
|
|
"""
|
|
db_contact_form = db.query(ContactForm).filter(ContactForm.id == contact_form_id).first()
|
|
if not db_contact_form:
|
|
return None
|
|
validate_contact_form_data(contact_form_data)
|
|
update_data = contact_form_data.dict(exclude_unset=True)
|
|
for key, value in update_data.items():
|
|
setattr(db_contact_form, key, value)
|
|
db.commit()
|
|
db.refresh(db_contact_form)
|
|
return db_contact_form
|
|
|
|
def validate_contact_form_data(contact_form_data: ContactFormCreate):
|
|
"""
|
|
Validates the contact form data.
|
|
|
|
Args:
|
|
contact_form_data (ContactFormCreate): The contact form data to validate.
|
|
|
|
Raises:
|
|
ValueError: If any of the validation rules are violated.
|
|
"""
|
|
if not contact_form_data.name or len(contact_form_data.name) < 3:
|
|
raise ValueError("Name must be at least 3 characters long.")
|
|
if not re.match(r"^[\w\.-]+@[\w\.-]+\.\w+$", contact_form_data.email):
|
|
raise ValueError("Invalid email address.")
|
|
if not contact_form_data.subject or len(contact_form_data.subject) < 5:
|
|
raise ValueError("Subject must be at least 5 characters long.")
|
|
if not contact_form_data.message or len(contact_form_data.message) < 10:
|
|
raise ValueError("Message must be at least 10 characters long.") |