Feature/email verification

.gitignore

@@ -1,3 +1,54 @@
-*.env
+# === Python build artifacts ===
 *.pyc
-__pycache__/
+*.pyo
+*.pyd
+__pycache__/
+*.log
+
+# === SQLite & output files ===
+*.sqlite3
+*.db
+output.json
+
+# === Environment variables ===
+.env
+.env.*
+*.env
+
+# === Virtual environments ===
+venv/
+.venv/
+.env/
+
+# === VSCode project settings ===
+.vscode/
+
+# === macOS system files ===
+.DS_Store
+
+# === Pytest and test cache ===
+htmlcov/
+.coverage
+.cache/
+pytest_cache/
+.tox/
+
+# === Jupyter Notebook ===
+.ipynb_checkpoints/
+
+# === Django migration artifacts (optional to ignore) ===
+# Uncomment the lines below if you want to regenerate migrations often
+# **/migrations/*.py
+# **/migrations/*.pyc
+# !**/migrations/__init__.py
+
+# === FastAPI-specific artifacts ===
+fastapi_email/email_db.sqlite3
+
+# === IDE-specific ===
+.idea/
+*.sublime-project
+*.sublime-workspace
+
+# === GitHub Codespaces or devcontainers ===
+.devcontainer/

fastapi_email/database.py

@@ -0,0 +1,15 @@
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker, declarative_base
+
+SQLALCHEMY_DATABASE_URL = "sqlite:///./email_db.sqlite3"
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL, connect_args={"check_same_thread": False})
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+Base = declarative_base()
+
+def get_db():
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

fastapi_email/main.py

@@ -0,0 +1,57 @@
+from fastapi import FastAPI, HTTPException, Depends
+from fastapi_email.schemas import EmailSchema
+from fastapi_email.database import engine, Base, get_db
+from fastapi_email.models import User
+from fastapi_email.utils import generate_token, send_verification_email, send_confirmation_email
+from sqlalchemy.orm import Session
+from datetime import datetime, timedelta
+
+Base.metadata.create_all(bind=engine)
+app = FastAPI()
+
+@app.post("/register")
+async def register(data: EmailSchema, db: Session = Depends(get_db)):
+    existing = db.query(User).filter(User.email == data.email).first()
+    if existing:
+        if existing.is_verified:
+            raise HTTPException(400, detail="Email already verified")
+        else:
+            existing.token = generate_token()
+            existing.token_expiry = datetime.utcnow() + timedelta(minutes=15)
+            db.commit()
+            await send_verification_email(existing.email, existing.token)
+            return {"message": "Verification email re-sent"}
+
+    token = generate_token()
+    expiry_time = datetime.utcnow() + timedelta(minutes=15)
+
+    new_user = User(email=data.email, token=token, token_expiry=expiry_time, is_verified=False)
+    db.add(new_user)
+    db.commit()
+    await send_verification_email(data.email, token)
+    return {"message": "Verification email sent"}
+
+
+@app.get("/verify/{token}")
+async def verify_email(token: str, db: Session = Depends(get_db)):
+    user = db.query(User).filter(User.token == token).first()
+    if not user:
+        raise HTTPException(400, detail="Invalid token")
+
+    if user.token_expiry and datetime.utcnow() > user.token_expiry:
+        raise HTTPException(400, detail="Verification link has expired")
+
+    user.is_verified = True
+    user.token = None
+    user.token_expiry = None
+    db.commit()
+    await send_confirmation_email(user.email)
+    return {"message": "Email verified"}
+
+
+@app.get("/status/{email}")
+def check_status(email: str, db: Session = Depends(get_db)):
+    user = db.query(User).filter(User.email == email).first()
+    if not user:
+        raise HTTPException(status_code=404, detail="User not found")
+    return {"email": email, "is_verified": user.is_verified}

fastapi_email/models.py

@@ -0,0 +1,10 @@
+from sqlalchemy import Column, String, Boolean, DateTime
+from fastapi_email.database import Base
+from datetime import datetime
+
+class User(Base):
+    __tablename__ = "users"
+    email = Column(String, primary_key=True, index=True)
+    token = Column(String, nullable=True)
+    token_expiry = Column(DateTime, nullable=True)
+    is_verified = Column(Boolean, default=False)

fastapi_email/schemas.py

@@ -0,0 +1,4 @@
+from pydantic import BaseModel, EmailStr
+
+class EmailSchema(BaseModel):
+    email: EmailStr

fastapi_email/utils.py

@@ -0,0 +1,43 @@
+import secrets
+from email.message import EmailMessage
+from aiosmtplib import send
+import os
+from dotenv import load_dotenv
+
+load_dotenv()
+
+def generate_token():
+    return secrets.token_urlsafe(32)
+
+async def send_verification_email(to_email, token):
+    msg = EmailMessage()
+    msg["Subject"] = "Verify your email"
+    msg["From"] = os.getenv("SMTP_FROM")
+    msg["To"] = to_email
+    link = f"{os.getenv('SITE_URL')}/verify/{token}"
+    msg.set_content(f"Click the link to verify your email:\n{link}")
+
+    await send(
+        msg,
+        hostname=os.getenv("SMTP_HOST"),
+        port=int(os.getenv("SMTP_PORT")),
+        username=os.getenv("SMTP_USER"),
+        password=os.getenv("SMTP_PASS"),
+        start_tls=True,
+    )
+
+async def send_confirmation_email(to_email):
+    msg = EmailMessage()
+    msg["Subject"] = "Email Successfully Verified"
+    msg["From"] = os.getenv("SMTP_FROM")
+    msg["To"] = to_email
+    msg.set_content("Thanks for verifying your email. You're all set!")
+
+    await send(
+        msg,
+        hostname=os.getenv("SMTP_HOST"),
+        port=int(os.getenv("SMTP_PORT")),
+        username=os.getenv("SMTP_USER"),
+        password=os.getenv("SMTP_PASS"),
+        start_tls=True,
+    )

requirements.txt

@@ -0,0 +1,10 @@
+fastapi==0.110.0
+uvicorn==0.29.0
+sqlalchemy==2.0.30
+aiosmtplib==2.0.2
+python-dotenv==1.0.1
+email-validator==2.1.1
+
+# For testing
+pytest==8.3.1
+pytest-mock==3.14.0

tests/test_main.py

@@ -0,0 +1,86 @@
+import pytest
+from fastapi.testclient import TestClient
+from fastapi_email.main import app
+from fastapi_email.database import Base, get_db, engine
+from fastapi_email.models import User
+from sqlalchemy.orm import sessionmaker
+from unittest.mock import patch
+
+# Use same engine as app, but with isolated session
+TestingSessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+def override_get_db():
+    try:
+        db = TestingSessionLocal()
+        yield db
+    finally:
+        db.close()
+
+# Override dependency
+app.dependency_overrides[get_db] = override_get_db
+client = TestClient(app)
+
+# ✅ Reset the database before each test
+@pytest.fixture(autouse=True)
+def reset_test_db():
+    Base.metadata.drop_all(bind=engine)
+    Base.metadata.create_all(bind=engine)
+
+
+@patch("fastapi_email.main.send_verification_email", return_value=None)
+@patch("fastapi_email.main.send_confirmation_email", return_value=None)
+def test_register_new_email(mock_confirm, mock_verify):
+    response = client.post("/register", json={"email": "abhi.playstore65@gmail.com"})
+    assert response.status_code == 200
+    assert "Verification email sent" in response.json()["message"]
+
+
+@patch("fastapi_email.main.send_verification_email", return_value=None)
+@patch("fastapi_email.main.send_confirmation_email", return_value=None)
+def test_duplicate_unverified_registration(mock_confirm, mock_verify):
+    client.post("/register", json={"email": "test.user2@example.com"})  # first call
+    response = client.post("/register", json={"email": "test.user2@example.com"})  # second call
+    assert response.status_code == 200
+    assert "Verification email re-sent" in response.json()["message"]
+
+
+def test_status_check_unverified():
+    client.post("/register", json={"email": "test.user3@example.com"})
+    response = client.get("/status/test.user3@example.com")
+    assert response.status_code == 200
+    assert response.json()["is_verified"] is False
+
+
+@patch("fastapi_email.main.send_verification_email", return_value=None)
+@patch("fastapi_email.main.send_confirmation_email", return_value=None)
+def test_verification_flow(mock_confirm, mock_verify):
+    client.post("/register", json={"email": "test.user4@example.com"})
+
+    # Access token from DB
+    db = next(override_get_db())
+    user = db.query(User).filter(User.email == "test.user4@example.com").first()
+    token = user.token
+
+    verify_response = client.get(f"/verify/{token}")
+    assert verify_response.status_code == 200
+    assert "Email verified" in verify_response.json()["message"]
+
+    # Confirm status
+    status_response = client.get("/status/test.user4@example.com")
+    assert status_response.status_code == 200
+    assert status_response.json()["is_verified"] is True
+
+
+@patch("fastapi_email.main.send_verification_email", return_value=None)
+@patch("fastapi_email.main.send_confirmation_email", return_value=None)
+def test_duplicate_verified_registration_fails(mock_confirm, mock_verify):
+    client.post("/register", json={"email": "test.user5@example.com"})
+
+    db = next(override_get_db())
+    user = db.query(User).filter(User.email == "test.user5@example.com").first()
+    user.is_verified = True
+    db.commit()
+
+    response = client.post("/register", json={"email": "test.user5@example.com"})
+    assert response.status_code == 400
+    assert response.json()["detail"] == "Email already verified"