← All samples

Lead-Capture Form Backend

A website quote form and its FastAPI + SQLite backend: validation, consent, honeypot spam blocking, per-IP rate limiting and a token-protected CSV export that's safe to open in Excel.

Run it

cd samples/lead-capture-api
pip install fastapi uvicorn
ADMIN_TOKEN=change-me uvicorn app:app
python test_app.py   # 13 tests

Files: app.py · form.html · test_app.py · README.md

Owner CSV export (from a real browser submission)

id,created_utc,name,email,phone,service,message
1,2026-10-19T13:19:50+00:00,Ava Patel,ava@example.com,,Website,"Hi, I need a 3-page website for my bakery."

Automated tests

PASS valid lead saved (201)
PASS bad email rejected (422)
PASS short message rejected (422)
PASS no consent rejected (422)
PASS honeypot submission accepted silently, nothing saved
PASS CSV export needs token (401)
PASS wrong token rejected (401)
PASS export has exactly the 2 real leads
PASS whitespace cleaned and email lower-cased
PASS formula injection neutralised in CSV
PASS empty optional field stays empty in CSV
PASS rate limit: 6th submission in 10 min gets 429
PASS form page served at /
13/13 tests passed

Code

"""Lead-capture backend for a small-business website contact form.
FastAPI + SQLite: validates input, blocks bots with a honeypot field, rate-limits per IP,
requires consent, and gives the owner a token-protected CSV export.
Run: ADMIN_TOKEN=change-me uvicorn app:app --reload   (needs fastapi, uvicorn)"""
import csv, io, os, re, sqlite3, time, datetime as dt
from collections import defaultdict, deque
from fastapi import FastAPI, HTTPException, Header, Request
from fastapi.responses import FileResponse, PlainTextResponse
from pydantic import BaseModel, Field, field_validator

DB = os.environ.get("LEADS_DB", "leads.db"); RATE = (5, 600)  # max 5 submissions per IP per 10 minutes
EMAIL = re.compile(r"^[^@\s]+@[^@\s]+\.[a-z]{2,}$", re.I)
app = FastAPI(title="Lead capture API (sample)"); hits = defaultdict(deque)

def db():
    c = sqlite3.connect(DB); c.execute("""CREATE TABLE IF NOT EXISTS leads(id INTEGER PRIMARY KEY, created TEXT, name TEXT, email TEXT,
        phone TEXT, service TEXT, message TEXT, source TEXT)"""); return c

class Lead(BaseModel):
    name: str = Field(min_length=2, max_length=80)
    email: str = Field(max_length=120)
    phone: str = Field(default="", max_length=30)
    service: str = Field(default="General question", max_length=60)
    message: str = Field(min_length=10, max_length=2000)
    consent: bool
    website: str = ""   # honeypot: hidden in the form, humans leave it empty
    @field_validator("email")
    @classmethod
    def email_ok(cls, v):
        v = v.strip().lower()
        if not EMAIL.match(v): raise ValueError("enter a valid email address")
        return v
    @field_validator("name", "message", "phone", "service")
    @classmethod
    def strip(cls, v): return " ".join(v.split())

@app.get("/")
def form(): return FileResponse(os.path.join(os.path.dirname(__file__), "form.html"))

@app.post("/api/leads", status_code=201)
def create(lead: Lead, request: Request):
    ip = request.client.host if request.client else "?"; now = time.time(); q = hits[ip]
    while q and now - q[0] > RATE[1]: q.popleft()
    if len(q) >= RATE[0]: raise HTTPException(429, "Too many submissions, please try again later.")
    q.append(now)
    if lead.website: return {"ok": True}  # silently drop bots
    if not lead.consent: raise HTTPException(422, "Please tick the consent box so we can reply.")
    with db() as c:
        cur = c.execute("INSERT INTO leads(created,name,email,phone,service,message,source) VALUES(?,?,?,?,?,?,?)",
                        (dt.datetime.now(dt.timezone.utc).isoformat(timespec="seconds"), lead.name, lead.email, lead.phone, lead.service, lead.message, request.headers.get("referer", "")))
    return {"ok": True, "id": cur.lastrowid}

@app.get("/api/leads.csv", response_class=PlainTextResponse)
def export(x_admin_token: str = Header(default="")):
    tok = os.environ.get("ADMIN_TOKEN", "")
    if not tok or x_admin_token != tok: raise HTTPException(401, "admin token required")
    with db() as c: rows = c.execute("SELECT id,created,name,email,phone,service,message FROM leads ORDER BY id").fetchall()
    buf = io.StringIO(); w = csv.writer(buf); w.writerow(["id", "created_utc", "name", "email", "phone", "service", "message"])
    for r in rows: w.writerow([("'" + v) if isinstance(v, str) and v and v[0] in "=+-@" else v for v in r])  # block spreadsheet formula injection
    return PlainTextResponse(buf.getvalue(), media_type="text/csv")