kin: BATON-008-backend_dev

backend/push.py

@@ -0,0 +1,35 @@
+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)