pywebpush

import os import json from pywebpush import webpush, WebPushException # --- Configuration --- # # Get VAPID private key from environment variable for security VAPID_PRIVATE_KEY = os.environ.get('WEBPUSH_VAPID_PRIVATE_KEY', 'your-vapid-private-key-here') # Your email or a mailto: URL for VAPID identification VAPID_SENDER_INFO = os.environ.get('WEBPUSH_SENDER_INFO', 'mailto:admin@example.com') # Example PushSubscription object (obtained from the client-side) # In a real application, this would be retrieved from a database. subscription_info = { "endpoint": "https://fcm.googleapis.com/fcm/send/SOME_ENDPOINT_ID", "keys": { "auth": "SOME_AUTH_KEY", "p256dh": "SOME_P256DH_KEY" } } # The payload data to send message_data = { "title": "Hello from pywebpush!", "body": "Your notification arrived.", "icon": "/images/notification-icon.png" } try: # Define VAPID claims. The 'aud' (audience) must be the origin of the push service endpoint. # pywebpush attempts to guess 'aud' but it's best to be explicit. # 'exp' (expiration) is set to 12 hours by default if not provided. vapid_claims = { "sub": VAPID_SENDER_INFO, "aud": subscription_info['endpoint'].split('/fcm/send/')[0] # Extract origin for FCM } print("Attempting to send push notification...") response = webpush( subscription_info=subscription_info, data=json.dumps(message_data), # Data should be a JSON string for common use cases vapid_private_key=VAPID_PRIVATE_KEY, vapid_claims=vapid_claims ) print(f"Push notification sent successfully! Status: {response.status_code}") print(f"Response: {response.text}") except WebPushException as e: print(f"Failed to send push notification: {e}") print(f"Response body: {e.response_body}") # Handle specific errors, e.g., expired subscription, invalid VAPID keys except Exception as e: print(f"An unexpected error occurred: {e}") # Example of using WebPusher class for more control (e.g., encoding data separately) # try: # pusher = WebPusher(subscription_info) # encoded_data = pusher.encode(json.dumps(message_data)) # # You can now send encoded_data with custom HTTP client if needed # # Or use pusher.send() if you want pywebpush to handle HTTP request # # response = pusher.send(json.dumps(message_data), headers=headers_dict) # headers_dict would include VAPID auth # print("Data encoded successfully via WebPusher.") # except WebPushException as e: # print(f"Error encoding data: {e}")