feat: Implement contact form submission endpoint with validation ✅ (auto-linted)

alembic/versions/20250428_134737_0265a0c8_update_contact_form.py

@@ -0,0 +1,29 @@
+"""add email_validator to contact_form table
+Revision ID: 6be7c5a3ec33
+Revises: 2b9e7c8a3f0e
+Create Date: 2023-05-25 14:23:12.921283
+"""
+from alembic import op
+import sqlalchemy as sa
+
+# revision identifiers, used by Alembic.
+revision = '6be7c5a3ec33'
+down_revision = '2b9e7c8a3f0e'
+branch_labels = None
+depends_on = None
+
+def upgrade():
+    table_name = "contact_forms"
+    # SQLite does not support altering columns, so we need to recreate the table
+    op.create_table(
+        table_name,
+        sa.Column('id', sa.String(36), primary_key=True, nullable=False),
+        sa.Column('name', sa.String(), nullable=False),
+        sa.Column('email', sa.String(), nullable=False),
+        sa.Column('message', sa.Text(), nullable=False),
+        sa.Column('created_at', sa.DateTime(), server_default=sa.func.now(), nullable=False),
+        sa.Column('updated_at', sa.DateTime(), server_default=sa.func.now(), nullable=False, onupdate=sa.func.now())
+    )
+
+def downgrade():
+    op.drop_table('contact_forms')

endpoints/form.post.py

@@ -1,25 +1,19 @@
 from fastapi import APIRouter, HTTPException, status
 from schemas.contact_form import ContactFormCreate
 from helpers.contact_form_helpers import create_contact_form, validate_email
-from sqlalchemy.orm import Session
 from fastapi import Depends
+from sqlalchemy.orm import Session
 from core.database import get_db
 
 router = APIRouter()
 
-@router.post("/form", status_code=status.HTTP_201_CREATED)
-async def create_contact_form_endpoint(
+@router.post("/form", status_code=status.HTTP_201_CREATED, response_model=ContactFormCreate)
+async def create_form(
     contact_form_data: ContactFormCreate,
     db: Session = Depends(get_db)
 ):
-    if not contact_form_data.name:
-        raise HTTPException(status_code=400, detail="Name is required")
-    if not contact_form_data.email:
-        raise HTTPException(status_code=400, detail="Email is required")
     if not validate_email(contact_form_data.email):
         raise HTTPException(status_code=400, detail="Invalid email format")
-    if not contact_form_data.message:
-        raise HTTPException(status_code=400, detail="Message is required")
 
     new_contact_form = create_contact_form(db=db, contact_form_data=contact_form_data)
     return new_contact_form

helpers/contact_form_helpers.py

@@ -2,7 +2,6 @@ from sqlalchemy.orm import Session
 from models.contact_form import ContactForm
 from schemas.contact_form import ContactFormCreate
 from fastapi import HTTPException
-import email_validator
 
 def validate_email(email: str) -> bool:
     """
@@ -14,11 +13,7 @@ def validate_email(email: str) -> bool:
     Returns:
         bool: True if the email is valid, False otherwise.
     """
-    try:
-        email_validator.validate_email(email)
-        return True
-    except email_validator.EmailNotValidError:
-        return False
+    return ContactForm.validate_email(email)
 
 def create_contact_form(db: Session, contact_form_data: ContactFormCreate) -> ContactForm:
     """
@@ -38,7 +33,7 @@ def create_contact_form(db: Session, contact_form_data: ContactFormCreate) -> Co
         raise HTTPException(status_code=400, detail="Name is required")
     if not contact_form_data.email:
         raise HTTPException(status_code=400, detail="Email is required")
-    if not validate_email(contact_form_data.email):
+    if not ContactForm.validate_email(contact_form_data.email):
         raise HTTPException(status_code=400, detail="Invalid email format")
     if not contact_form_data.message:
         raise HTTPException(status_code=400, detail="Message is required")

models/contact_form.py

@@ -3,6 +3,7 @@ from sqlalchemy.dialects.postgresql import UUID
 from sqlalchemy.sql import func
 from core.database import Base
 import uuid
+import email_validator
 
 class ContactForm(Base):
     __tablename__ = "contact_forms"
@@ -13,3 +14,20 @@ class ContactForm(Base):
     message = Column(Text, nullable=False)
     created_at = Column(DateTime, default=func.now())
     updated_at = Column(DateTime, default=func.now(), onupdate=func.now())
+
+    @staticmethod
+    def validate_email(email: str) -> bool:
+        """
+        Validates the provided email address using the email_validator package.
+
+        Args:
+            email (str): The email address to validate.
+
+        Returns:
+            bool: True if the email is valid, False otherwise.
+        """
+        try:
+            email_validator.validate_email(email)
+            return True
+        except email_validator.EmailNotValidError:
+            return False

schemas/contact_form.py

@@ -1,19 +1,26 @@
-from pydantic import BaseModel, EmailStr, Field
+import uuid
+from pydantic import BaseModel, Field, validator
 from typing import Optional
 from datetime import datetime
-import uuid
+import email_validator
 
 class ContactFormBase(BaseModel):
     name: str = Field(..., description="Contact name")
-    email: EmailStr = Field(..., description="Contact email address")
+    email: str = Field(..., description="Contact email address")
     message: str = Field(..., description="Contact message")
 
+    @validator('email')
+    def validate_email(cls, value):
+        if not email_validator.validate_email(value):
+            raise ValueError('Invalid email address')
+        return value
+
 class ContactFormCreate(ContactFormBase):
     pass
 
 class ContactFormUpdate(ContactFormBase):
     name: Optional[str] = Field(None, description="Contact name")
-    email: Optional[EmailStr] = Field(None, description="Contact email address")
+    email: Optional[str] = Field(None, description="Contact email address")
     message: Optional[str] = Field(None, description="Contact message")
 
 class ContactFormSchema(ContactFormBase):