Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
How to Recieive a webhook from Chargify to a Flask API endpoint
from flask import Flask, request, Response, abort
from flask.ext.restful import Api, Resource,
from hashlib import sha256
import hmac, redis
CHARGIFY_SHARED_KEY= "XYZ789"
REDIS_HOST="ip_address"
REDIS_PORT="1234"
FLASK_HOST="ip_address"
FLASK_PORT="1234"
app = Flask(__name__)
api = Api(app)
r = redis.StrictRedis(host=REDIS_HOST, port=REDIS_PORT, db=1)
class Customer_Signup(Resource):
def parse_chargify_webhook(post_data):
'''
Converts Chargify webhook parameters to a python dictionary of nested dictionaries
'''
result = {}
for k, v in post_data.iteritems():
keys = [x.strip(']') for x in k.split('[')]
cur = result
for key in keys[:-1]:
cur = cur.setdefault(key, {})
cur[keys[-1]] = v
return result
def post(self):
'''
Recives a Chargify sends a webhook
(https://docs.chargify.com/webhooks#signup-success-payload)
'''
# 1 Compare the Chargify webhook signature, if not then abort
signature = request.headers.get('X-Chargify-Webhook-Signature-Hmac-Sha-256')
if not hmac.compare_digest(signature, hmac.new(CHARGIFY_SHARED_KEY, request.get_data(), sha256).hexdigest()): abort(404)
# Python 2.7.6 and lower, can use this less secure version: if not (signature == hmac.new(CHARGIFY_SHARED_KEY, request.get_data(), sha256).hexdigest()): abort(404)
# 2. Check the webhook id has not been processed already
# 2.1 Get the webhook_id of this webhook
webhook_data=parse_chargify_webhook(request.form)
webhook_id=webhook_data["id"]
# 2.1 If the key does not exisit in redis, set a new key
if (r.get(webhook_id) == None):
# key to expire after 330 seconds (chargify limit for resent webhooks)
r.setex(webhook_id, 330, 'completed')
# continue
else: # this key exists and does not need to be processed again, so exit
return Response(status=202)
# 3 Get the customer_id from the webhook payload
customer_id=webhook_data["payload"]["subscription"]["customer"]["id"]
# 4. Call the main function to be executed (need to use celery for long running tasks)
customer_signup_function(customer_id)
# 5. Tell chargify we have processed this webhook correctly (within 10 seconds)
return Response(status=202)
api.add_resource(Customer_Signup, '/api/customer_signup/')
if __name__ == '__main__':
app.run(host=FLASK_HOST, port=FLASK_PORT)
@robschaap242
Copy link

robschaap242 commented Jan 13, 2017

Thanks man, going to try this out today or tomorrow, the python docs / SDKS they provide are ... not up to date or missing. This is the nuts and bolts of anything - well organized and presented in your blog post too!

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment