35 lines
1.1 KiB
Python
35 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import json
|
|
import logging
|
|
|
|
from backend import config
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
async def send_push(subscription_json: str, title: str, body: str) -> None:
|
|
"""Send a Web Push notification. Swallows all errors — never raises."""
|
|
if not config.VAPID_PRIVATE_KEY:
|
|
logger.warning("VAPID_PRIVATE_KEY not configured — push notification skipped")
|
|
return
|
|
try:
|
|
import pywebpush # type: ignore[import]
|
|
except ImportError:
|
|
logger.warning("pywebpush not installed — push notification skipped")
|
|
return
|
|
try:
|
|
subscription_info = json.loads(subscription_json)
|
|
data = json.dumps({"title": title, "body": body})
|
|
vapid_claims = {"sub": f"mailto:{config.VAPID_CLAIMS_EMAIL or 'admin@example.com'}"}
|
|
|
|
await asyncio.to_thread(
|
|
pywebpush.webpush,
|
|
subscription_info=subscription_info,
|
|
data=data,
|
|
vapid_private_key=config.VAPID_PRIVATE_KEY,
|
|
vapid_claims=vapid_claims,
|
|
)
|
|
except Exception as exc:
|
|
logger.error("Web Push failed: %s", exc)
|