Send email from FastAPI

import logging
from contextlib import asynccontextmanager

import httpx
from fastapi import BackgroundTasks, FastAPI, HTTPException
from pydantic import BaseModel, EmailStr

from settings import settings

log = logging.getLogger("uvicorn.error")


@asynccontextmanager
async def lifespan(app: FastAPI):
    # One client for the whole process. Creating an AsyncClient per request
    # throws away connection pooling and the TLS handshake every time.
    app.state.http = httpx.AsyncClient(
        base_url="https://emails.sh/v1",
        headers={"Authorization": f"Bearer {settings.emailssh_api_key}"},
        timeout=15.0,
    )
    try:
        yield
    finally:
        await app.state.http.aclose()


app = FastAPI(lifespan=lifespan)


class SignupRequest(BaseModel):
    email: EmailStr
    name: str = "there"


async def send_welcome(client: httpx.AsyncClient, email: str, name: str) -> None:
    response = await client.post(
        "/emails",
        json={
            "from": settings.emailssh_from,
            "to": [email],
            "subject": "Welcome to Acme",
            "html": f"<p>Hi {name}, your Acme account is ready.</p>",
            "text": f"Hi {name}, your Acme account is ready.",
            "idempotency_key": f"welcome:{email.lower()}",
        },
    )

    if response.is_error:
        # The body is prose and says what to do next, so log all of it.
        log.error("emails.sh %s: %s", response.status_code, response.text)
        return

    log.info("emails.sh queued %s", response.json()["id"])


@app.post("/signup")
async def signup(payload: SignupRequest, background_tasks: BackgroundTasks):
    if not settings.emailssh_api_key:
        raise HTTPException(status_code=500, detail="Email is not configured")

    background_tasks.add_task(send_welcome, app.state.http, payload.email, payload.name)
    return {"ok": True}

import hashlib
import hmac
import os

from fastapi import APIRouter, Header, HTTPException, Request

router = APIRouter()
SECRET = os.environ["EMAILSSH_WEBHOOK_SECRET"].encode()


@router.post("/webhooks/emailssh", status_code=204)
async def emailssh_events(request: Request, x_emailssh_signature: str = Header(default="")):
    # Read the raw body, not the parsed JSON. Re-serialising changes the bytes
    # and the signature will never match.
    raw = await request.body()
    expected = hmac.new(SECRET, raw, hashlib.sha256).hexdigest()

    if not hmac.compare_digest(expected, x_emailssh_signature):
        raise HTTPException(status_code=401, detail="bad signature")

    event = await request.json()

    if event["type"] in ("email.bounced", "email.complained"):
        # Mark the address unsendable here.
        pass

    return None