Store SMS delivery receipts with FastAPI

import json
import os
import sqlite3
from datetime import datetime

from bird import Bird, WebhookVerificationError
from fastapi import FastAPI, HTTPException, Request, Response

app = FastAPI()
DB = os.environ.get("SMS_EVENTS_DB", "sms-events.sqlite3")
STATUSES = {
    "sms.accepted": "accepted", "sms.sent": "sent",
    "sms.delivered": "delivered", "sms.undelivered": "undelivered",
    "sms.failed": "failed", "sms.expired": "expired", "sms.rejected": "rejected",
}


def database():
    db = sqlite3.connect(DB, timeout=5)
    db.execute("CREATE TABLE IF NOT EXISTS events (id TEXT PRIMARY KEY, body TEXT NOT NULL)")
    db.execute("""CREATE TABLE IF NOT EXISTS outcomes (
        workspace_id TEXT, sms_id TEXT, status TEXT, observed_at REAL,
        PRIMARY KEY (workspace_id, sms_id))""")
    db.commit()
    return db


@app.post("/webhooks/bird")
async def receive(request: Request):
    raw = await request.body()
    try:
        with Bird(webhook_secret=os.environ["BIRD_WEBHOOK_SECRET"]) as bird:
            event = bird.webhooks.unwrap(raw, dict(request.headers)).root
    except WebhookVerificationError:
        raise HTTPException(400, "Invalid webhook signature or payload.")
    event_id = request.headers["webhook-id"]
    body = event.model_dump(mode="json", by_alias=True)
    if body["type"] not in STATUSES and body["type"] != "sms.received":
        return Response(status_code=204)
    data = body["data"]
    observed_at = datetime.fromisoformat(body["timestamp"].replace("Z", "+00:00")).timestamp()
    db = database()
    try:
        with db:
            added = db.execute("INSERT OR IGNORE INTO events VALUES (?, ?)",
                               (event_id, json.dumps(body))).rowcount
            if added and body["type"] in STATUSES:
                db.execute("""INSERT INTO outcomes VALUES (?, ?, ?, ?)
                    ON CONFLICT(workspace_id, sms_id) DO UPDATE SET
                      status=excluded.status, observed_at=excluded.observed_at
                    WHERE excluded.observed_at > outcomes.observed_at""",
                    (data["workspace_id"], data["sms_id"], STATUSES[body["type"]], observed_at))
    finally:
        db.close()
    return Response(status_code=204)

python -c 'import sqlite3; db=sqlite3.connect("sms-events.sqlite3"); print(db.execute("SELECT sms_id, status FROM outcomes").fetchall()); db.close()'