README.md

Backend web push service

To run the API, following settings need to be exported as env variables

• VAPID email
• VAPID private key path
• VAPID public key path

Running the API

$ python api.py

API Requests

$ curl -X GET http://127.0.0.1:5000/subscription
$ curl -X POST http://127.0.0.1:5000/subscription/ -d '{subsctption_token:token}'
$ curl -X POST http://127.0.0.1:5000/push

api.py

import logging
import json

from flask import request, Response
from flask import jsonify

from flask_api import FlaskAPI
from flask_sqlalchemy import SQLAlchemy

from settings import VAPID_PUBLIC_KEY
from push import send_web_push

app = FlaskAPI(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:////tmp/test.db'
db = SQLAlchemy(app)

logger = logging.getLogger(__name__)

class User(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    subscription_token = db.Column(db.String(400), unique=True)
    
    def __init__(self, subscription_token):
        self.subscription_token = subscription_token
        
    def __repr__(self):
        return "<User %r>" % self.id
    
@app.route("/subscription/", methods=["GET", "POST"])
def subscription():
    """
        POST creates a subscription
        GET returns vapid public key which clients uses to send around push notification
    """
    
    if request.method == "GET":
        return Response(response=json.dumps({"public_key": VAPID_PUBLIC_KEY}), 
            headers={"Access-Control-Allow-Origin": "*"}, content_type="application/json")
    
    subscription_token = request.get_json("subscription_token")
    return Response(status=201, mimetype="application/json")
    
@app.route("/push/", methods=["POST"])
def push_to_all_users():
    message = request.get_json("message") if request.data else "Updates available"
        
    for user in User.query.all():
        try:
            send_web_push(json.loads(user.subscription_token), message)
        except Exception as e:
            logger.error(e)
            
    return Response(status=200, mimetype="application/json")
        
        
if __name__=="__main__":
    app.run(debug=True)

push.py

import logging

from settings import *
from pywebpush import webpush, WebPushException


logger = logging.getLogger(__name__)

def send_web_push(subscription_information, message_body):
    try:
        webpush(
            subscription_info=subscription_information,
            data=message_body,
            vapid_private_key=VAPID_PRIVATE_KEY,
            vapid_claims=VAPID_CLAIMS
        )
    except WebPushException as exception:
        logger.error(exception)

requirements.txt

Flask-API==1.0
Flask-SQLAlchemy==2.2
py-vapid==1.2.5
cryptography==2.0.3 
pywebpush==1.0.6