py-push-server/app.py

from flask import Flask, request, jsonify
from models import db, VAPIDKey, Subscription
from py_vapid import Vapid
from pywebpush import webpush, WebPushException

import json
import os

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///webpush.db'
db.init_app(app)

def generate_and_save_vapid_keys():
    vapid = Vapid()
    vapid.generate_keys()
    private_key = vapid.get_private_key().to_pem().decode('utf-8').strip()
    public_key = vapid.get_public_key().to_pem().decode('utf-8').strip()

    key = VAPIDKey(public_key=public_key, private_key=private_key)
    db.session.add(key)
    db.session.commit()

    
@app.before_first_request
def initialize():
    if not VAPIDKey.query.first():
        generate_and_save_vapid_keys()
    
def send_push_notification(subscription_info, message, vapid_key):
    try:
        webpush(
            subscription_info=subscription_info,
            data=json.dumps(message),
            vapid_private_key=vapid_key.private_key,
            vapid_claims={
                "sub": "mailto:your-email@example.com"
            }
        )
    except WebPushException as e:
        print(f"Failed to send notification: {e}")

        
@app.route('/web-push/vapid', methods=['GET'])
def get_vapid():
    key = VAPIDKey.query.first()
    if key:
        return jsonify(public_key=key.public_key)
    return jsonify(error='No VAPID keys found'), 404


@app.route('/web-push/subscribe', methods=['POST'])
def subscribe():
    content = request.json
    vapid_key = VAPIDKey.query.first()
    
    if not vapid_key:
        return jsonify(success=False, error="No VAPID keys available"), 500

    subscription = Subscription(endpoint=content['endpoint'],
                                p256dh=content['keys']['p256dh'],
                                auth=content['keys']['auth'],
                                vapid_key_id=vapid_key.id)
    db.session.add(subscription)
    db.session.commit()

    subscription_info = {
        "endpoint": subscription.endpoint,
        "keys": {
            "p256dh": subscription.p256dh,
            "auth": subscription.auth
        }
    }

    message = {"title": "Subscription Successful", "body": "Thank you for subscribing!"}
    send_push_notification(subscription_info, message, vapid_key)

    return jsonify(success=True)


@app.route('/web-push/unsubscribe', methods=['DELETE'])
def unsubscribe():
    content = request.json
    endpoint = content['endpoint']
    subscription = Subscription.query.filter_by(endpoint=endpoint).first()

    if subscription:
        db.session.delete(subscription)
        db.session.commit()
        return jsonify(success=True, message="Subscription deleted successfully")
    else:
        return jsonify(success=False, error="Subscription not found"), 404

    
if __name__ == '__main__':
    app.run(debug=True)