Develop Web Push Notification Server With Python + Flask

Web push libraries are used to send out push notifications (available for Java, C#, C, Node.js, PHP and Python).

We shall develop a Web Push Notification Server (e.g. https://web-push-codelab.glitch.me/) with Python and Flask.

NOTE: you might want to read about Setup Push Notification For PWA (client side).

VAPID public key and private key

Before setup a Push Notification Server, we need generate VAPID public key and private key.

sudo pip install py-vapid

Create a claim.json file.

sub: is the email address you wish to have on record for this request, prefixed with mailto:. If things go wrong, this is the email that will be used to contact you (for instance). This can be a general delivery address like mailto:[email protected] or a specific address like mailto:[email protected].
aud: is the audience for the VAPID. This is the scheme and host you use to send subscription endpoints and generally coincides with the endpoint specified in the Subscription Info block.

{"sub":"mailto:[email protected]","aud":"https://www.mydomain.com"}

The following command will generate private_key.pem and public_key.pem.

vapid --sign claim.json

Run this command to generate VAPID public key.

vapid --applicationServerKey

Run this command to generate VAPID private key.

openssl ec -in private_key.pem -outform DER|tail -c +8|head -c 32|base64|tr -d '=' |tr '/+' '_-'

NOTE: there seems to be an easier way to generate VAPID keys in Node.js using web-push-libs/web-push by using webpush.generateVAPIDKeys(). You can also generate VAPID keys using openssl.

pywebpush

Install pywebpush.

pip install pywebpush

NOTE: I bump into the following error Collecting cryptography==2.1.4 (from -r requirements.txt (line 1)) Using cached cryptography-2.1.4.tar.gz Complete output from command python setup.py egg_info: error in cryptography setup command: Invalid environment marker: platform_python_implementation != 'PyPy' due to lower pip version You are using pip version 7.1.0, however version 9.0.1 is available.. Just upgrade pip pip install --upgrade pip and the problem is solved.

To send a push notification is pretty easy.

from pywebpush import webpush, WebPushExceptionimport loggingWEBPUSH_VAPID_PRIVATE_KEY = 'xxx'# from PushManager: https://developer.mozilla.org/en-US/docs/Web/API/PushManagersubscription_info = {"endpoint": "https://updates.push.services.mozilla.com/push/v1/gAA...", "keys": {"auth": "k8J...", "p256dh": "BOr..."}}try:    webpush(        subscription_info=subscription_info,        data="Test 123", # could be json object as well        vapid_private_key=WEBPUSH_VAPID_PRIVATE_KEY,        vapid_claims={            "sub": "mailto:[email protected]"        }    )    count += 1except WebPushException as e:    logging.exception("webpush fail")

Flask Web Push Notification Server

We create a SQLAlchemy model to store web push subscription.

import jsonfrom flask import Flaskfrom sqlalchemy import Column, Integer, String, Text, DateTime, Booleanapp = Flask(__name__)db = SQLAlchemy(app)# db is SQLAlchemy objectclass Subscriber(db.Model):    __tablename__ = 'subscriber'    id = Column(Integer(), primary_key=True, default=None)    created = Column(DateTime())    modified = Column(DateTime())    subscription_info = Column(Text())    is_active = Column(Boolean(), default=True)    @property    def subscription_info_json(self):        return json.loads(self.subscription_info)    @subscription_info_json.setter    def subscription_info_json(self, value):        self.subscription_info = json.dumps(value)

We create a Flask REST API to accept subscription.

import datetimefrom flask import jsonify, request@app.route('/api/subscribe')def subscribe():    subscription_info = request.json.get('subscription_info')    # if is_active=False == unsubscribe    is_active = request.json.get('is_active')    # we assume subscription_info shall be the same    item = Subscriber.query.filter(Subscriber.subscription_info == subscription_info).first()    if not item:        item = Subscriber()        item.created = datetime.datetime.utcnow()        item.subscription_info = subscription_info    item.is_active = is_active    item.modified = datetime.datetime.utcnow()    db.session.add(item)    db.session.commit()    return jsonify({ id: item.id })

We create a function to send push notification to all subscribers.

@app.route('/notify')def notify():    from pywebpush import webpush, WebPushException    WEBPUSH_VAPID_PRIVATE_KEY = 'xxx'    items = Subscriber.query.filter(Subscriber.is_active == True).all()    count = 0    for _item in items:        try:            webpush(                subscription_info=_item.subscription_info_json,                data="Test 123",                vapid_private_key=WEBPUSH_VAPID_PRIVATE_KEY,                vapid_claims={                    "sub": "mailto:[email protected]"                }            )            count += 1        except WebPushException as ex:            logging.exception("webpush fail")    return "{} notification(s) sent".format(count)

References: