Develop Web Push Notification Server With Python + Flask

February 13, 2018

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.

Install py-vapid.

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:push_operations@example.com or a specific address like mailto:bob@example.com.
  • 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:webpush@mydomain.com","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, WebPushException
import logging

WEBPUSH_VAPID_PRIVATE_KEY = 'xxx'

# from PushManager: https://developer.mozilla.org/en-US/docs/Web/API/PushManager
subscription_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:webpush@mydomain.com"
        }
    )
    count += 1
except WebPushException as e:
    logging.exception("webpush fail")

Flask Web Push Notification Server

We create a SQLAlchemy model to store web push subscription.

import json
from flask import Flask
from sqlalchemy import Column, Integer, String, Text, DateTime, Boolean

app = Flask(__name__)
db = SQLAlchemy(app)

# db is SQLAlchemy object
class 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 datetime
from 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:webpush@mydomain.com"
                }
            )
            count += 1
        except WebPushException as ex:
            logging.exception("webpush fail")


    return "{} notification(s) sent".format(count)

References:

This work is licensed under a
Creative Commons Attribution-NonCommercial 4.0 International License.