app.py

# pip install fastapi[standard] webpush

from pathlib import Path
import uuid
from typing import Annotated
import datetime

from webpush import WebPush, WebPushSubscription
import httpx

from pydantic import BaseModel
from fastapi import FastAPI, Request, BackgroundTasks, Response, Cookie
from fastapi.templating import Jinja2Templates
from fastapi.staticfiles import StaticFiles

class Message(BaseModel):
    title: str
    body: str
    require_interaction: bool = False
    icon: str # URL
    #tag: str
    renotify: bool = False
    timestamp: float

templates = Jinja2Templates(directory="templates")
wp = WebPush(
    private_key=Path("./private_key.pem"),
    public_key=Path("./public_key.pem"),
    subscriber="eskildsf@gmail.com",
    ttl=30,
)
subscriptions: dict[str, WebPushSubscription] = {}
app = FastAPI()

@app.get("/")
async def root(request: Request):
    return templates.TemplateResponse(
        "index.html",
        context={
            "request": request,
            "application_server_key": "BNaVxIyIyiTJrnjkeA_58YU-9nEgu3fCvBYFl44VA3wTTgygalXyXJvmGgwU4etN1nTQwTZiCNmqdCRy4B8H3j0",
        }
    )

@app.post("/subscribe")
async def subscribe_user(
        subscription: WebPushSubscription, 
        background_tasks: BackgroundTasks,
        response: Response):
    id = str(uuid.uuid4())
    response.set_cookie(key="session", value=id)
    subscriptions[id] = subscription

@app.get("/notify")
async def notify(
        session: Annotated[str, Cookie()],
        background_tasks: BackgroundTasks,
        request: Request):
    assert session in subscriptions
    subscription = subscriptions[session]
    message = wp.get(
        message=Message(
            title="TFF410A",
            body=f"Watch activated: T > 30 degC\n{datetime.datetime.now():%y-%m-%d %H:%M:%S}",
            icon=str(request.url_for("static", path="favicon.ico")),
            #tag="TFF410A",
            timestamp=datetime.datetime.now().timestamp(),
        ).model_dump_json(),
        subscription=subscription,
    )
    background_tasks.add_task(
        httpx.post,
        url=str(subscription.endpoint),
        data=message.encrypted,
        headers=message.headers
    )

@app.get("/notify_all")
async def notify(
        background_tasks: BackgroundTasks,
        request: Request):
    for subscription in subscriptions.values():
        message = wp.get(
            message=Message(
                title="TFF410A",
                body=f"Watch activated: T > 30 degC\n{datetime.datetime.now():%y-%m-%d %H:%M:%S}",
                icon=str(request.url_for("static", path="favicon.ico")),
                #tag="TFF410A",
                timestamp=datetime.datetime.now().timestamp(),
            ).model_dump_json(),
            subscription=subscription,
        )
        background_tasks.add_task(
            httpx.post,
            url=str(subscription.endpoint),
            data=message.encrypted,
            headers=message.headers
        )

@app.get("/app.webmanifest")
async def pwa():
    return {"name": "Open Pectus", "id": "12345", "display": "standalone",}

app.mount("/", StaticFiles(directory="static"), name="static")