Create a gist now

Instantly share code, notes, and snippets.

@mushfiq /README.md
Last active Sep 24, 2017

What would you like to do?

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
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)
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)
Flask-API==1.0
Flask-SQLAlchemy==2.2
py-vapid==1.2.5
cryptography==2.0.3
pywebpush==1.0.6
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment