FastAPI

pip install fastapi uvicorn httpx

# .env, or set it in the environment of whatever runs uvicorn.
# The key comes from https://emails.sh/dashboard/api-keys.
EMAILSSH_API_KEY=esh_your_key_here

# main.py
import logging
import os
from contextlib import asynccontextmanager

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

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


@asynccontextmanager
async def lifespan(app: FastAPI):
    # One client for the process: a new one per request leaks connections and
    # loses the pool, which shows up as latency long before it shows up as an error.
    app.state.http = httpx.AsyncClient(
        base_url="https://emails.sh",
        headers={"Authorization": f"Bearer {os.environ['EMAILSSH_API_KEY']}"},
        timeout=10.0,
    )
    yield
    await app.state.http.aclose()


app = FastAPI(lifespan=lifespan)


class Signup(BaseModel):
    email: EmailStr


async def send_welcome(client: httpx.AsyncClient, address: str) -> None:
    response = await client.post(
        "/v1/emails",
        json={
            "from": "Acme <hello@acme.com>",
            "to": [address],
            "subject": "Welcome to Acme",
            "html": "<p>Confirm your address to finish signing up.</p>",
            "idempotency_key": f"signup-{address}",
        },
    )
    if response.status_code >= 400:
        refusal = response.json()["error"]
        log.error("emails.sh refused the send: %s: %s", refusal["code"], refusal.get("next", refusal["message"]))


@app.post("/signup")
async def signup(body: Signup, background: BackgroundTasks):
    # Answer now, send after the response has gone out.
    background.add_task(send_welcome, app.state.http, body.email)
    return {"accepted": True}


@app.post("/signup-sync")
async def signup_sync(body: Signup):
    response = await app.state.http.post(
        "/v1/emails",
        json={
            "from": "Acme <hello@acme.com>",
            "to": [body.email],
            "subject": "Welcome to Acme",
            "html": "<p>Confirm your address to finish signing up.</p>",
        },
    )
    if response.status_code >= 400:
        refusal = response.json()["error"]
        log.error("emails.sh refused the send: %s: %s", refusal["code"], refusal["message"])
        raise HTTPException(status_code=502, detail="Could not send the confirmation email.")

    return {"id": response.json()["id"]}

uvicorn main:app --reload