Web Push Notifications Integration Guide¶

from homepot.push_notifications import get_push_provider

# Configure Web Push provider
config = {
    "vapid_private_key": os.getenv("WEB_PUSH_VAPID_PRIVATE_KEY"),
    "vapid_public_key": os.getenv("WEB_PUSH_VAPID_PUBLIC_KEY"),
    "vapid_subject": os.getenv("WEB_PUSH_VAPID_SUBJECT"),
    "ttl_seconds": int(os.getenv("WEB_PUSH_TTL_SECONDS", 300)),
    "timeout_seconds": int(os.getenv("WEB_PUSH_TIMEOUT_SECONDS", 30)),
}

# Get Web Push provider
web_push_provider = await get_push_provider("web_push", config)

from fastapi import APIRouter, HTTPException
from pydantic import BaseModel
import json

router = APIRouter(prefix="/push", tags=["Push Notifications"])

class PushSubscription(BaseModel):
    subscription: dict
    device_info: dict = {}

@router.get("/vapid-public-key")
async def get_vapid_public_key():
    """Get VAPID public key for client-side subscription."""
    try:
        provider = await get_push_provider("web_push")
        public_key = provider.get_vapid_public_key()

        return {
            "publicKey": public_key,
            "platform": "web_push"
        }
    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Failed to get VAPID public key: {str(e)}"
        )

@router.post("/subscribe")
async def subscribe_to_push(subscription_data: PushSubscription):
    """Store push subscription for a user/device."""
    try:
        # Store subscription in database
        device_token = json.dumps(subscription_data.subscription)

        # Validate subscription format
        provider = await get_push_provider("web_push")
        is_valid = provider.validate_device_token(device_token)

        if not is_valid:
            raise HTTPException(
                status_code=400,
                detail="Invalid subscription format"
            )

        # TODO: Store in database with user/device association
        # await db.save_push_subscription(
        #     user_id=current_user.id,
        #     platform="web_push",
        #     device_token=device_token,
        #     device_info=subscription_data.device_info
        # )

        return {
            "status": "success",
            "message": "Push subscription registered",
            "platform": "web_push"
        }

    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Failed to register subscription: {str(e)}"
        )

@router.post("/send")
async def send_push_notification(
    device_token: str,
    title: str,
    body: str,
    data: dict = {},
    icon: str = None,
):
    """Send a test push notification."""
    try:
        provider = await get_push_provider("web_push")

        payload = PushNotificationPayload(
            title=title,
            body=body,
            data=data,
            priority=PushPriority.HIGH,
            platform_data={
                "icon": icon or "/icon-192.png",
                "badge": "/badge-72.png",
            }
        )

        result = await provider.send_notification(device_token, payload)

        return {
            "status": "success" if result.success else "failed",
            "message": result.message,
            "platform": result.platform
        }

    except Exception as e:
        raise HTTPException(
            status_code=500,
            detail=f"Failed to send push notification: {str(e)}"
        )

from homepot.push_notifications import get_push_provider, PushNotificationPayload

# Get provider
provider = await get_push_provider("web_push")

# Device token is the subscription object as JSON
device_token = json.dumps({
    "endpoint": "https://fcm.googleapis.com/fcm/send/...",
    "keys": {
        "p256dh": "BNcR...base64url...",
        "auth": "tBHI...base64url..."
    }
})

# Create payload
payload = PushNotificationPayload(
    title="Payment Update",
    body="New payment gateway configuration available",
    data={
        "url": "/config/payment",
        "version": "2.1.0"
    },
    priority=PushPriority.HIGH,
    platform_data={
        "icon": "/icons/payment.png",
        "badge": "/icons/badge.png",
        "requireInteraction": True,
        "actions": [
            {"action": "view", "title": "View Details"},
            {"action": "dismiss", "title": "Dismiss"}
        ]
    }
)

# Send notification
result = await provider.send_notification(device_token, payload)

if result.success:
    print(f"✓ Notification sent: {result.message}")
else:
    print(f"✗ Failed: {result.message} ({result.error_code})")

# Prepare notifications for multiple browsers
notifications = [
    (subscription1_json, payload1),
    (subscription2_json, payload2),
    (subscription3_json, payload3),
]

# Send all notifications
results = await provider.send_bulk_notifications(notifications)

# Check results
success_count = sum(1 for r in results if r.success)
print(f"Sent {success_count}/{len(results)} notifications successfully")

# Use simulation provider for testing without real credentials
sim_provider = await get_push_provider("simulation")

payload = PushNotificationPayload(
    title="Test Notification",
    body="This is a test",
)

result = await sim_provider.send_notification("test-token", payload)
print(f"Simulation result: {result.message}")

import logging

logging.basicConfig(level=logging.DEBUG)
logger = logging.getLogger("homepot.push_notifications.web_push")
logger.setLevel(logging.DEBUG)