What is Webhook ?
Webhook is an asynchronous HTTP callback on an event occurrence. It is a simple server to server communication for reporting a specific event occurred on a server. The server on which event occurred will fire a HTTP POST request to another server on a URL which is provided by receiving server.
For example, whenever your colleague pushes code commits to github, an event has occurred on github’s server. Now if a webhook URL is provided in github settings, a webhook will be fired to that URL. This webhook will be a HTTP POST request with commit details inside the body in a specified format. More details on github webhook can be found here.
In this post, I will share my experience of implementing webhook handler in python. For the readers, basic knowledge on implementing web application in python would be better.
A Webhook can be handled by simply providing a URL endpoint in a web application. Following is an example using Django. Add webhook url in urls.py
from django.conf.urls import url import views urlpatterns = [ url(r'^webhook', views.webhook, name='webhook'), ]
Now create view function in views.py which will parse the data and process it. In most of the cases, webhook data is sent in JSON format. So lets load the webhook data and sent the data to process_webhook function.
Most of the web applications accept POST request after verifying CSRF token, but here we need to exempt it from this check. So put @csrf_token decorator above the view function. Also put an @require_post decorator to ensure the request is only POST.
from django.views.decorators.csrf import csrf_exempt from django.views.decorators.http import require_POST @require_POST @csrf_exempt def webhook(request): # Load the event data from JSON data = json.loads(request.body) # And process it process_webhook(data) return 200, 'Processed.'
The above implementation of URL endpoint will remain different for various other python web framework like Flask, tornado, twisted. But the below code process_webhook function implementation will remain same irrespective of any framework.
There may be different type events we need to handle. So, before proceeding to implement process_webhook function, lets create a python module named webhook_events.py, which will contain a single function for each type of event wherein will be the logic for that particular event. In other words, we are going to map event name with its function, which will handle the logic for that particular type of webhook event.
def event_one(event): # do something for # for event 'event.one' def event_two(event): # do something for # for event 'event.two'
There are many ways to implement process_webhook function and how we map a webhook event with its function. We are going to discuss different implementation of process_webhook based on extendability. Most basic version of that is below.
import webhook_events def process_webhook(event): event_name = event['name'] if event_name == 'event.one': webhook_event.event_one(event) elif event_name == 'event.two': webhook_event.event_two(event) # and so on
A Better way
Now suppose, there are 10s of webhook to be served. We certainly don’t want to write repetitive code. So below is a better way of implementing process_webhook. Here we just replace dot in event name with underscore, so that we get the function name written in webhook_events.py for that event. If the function is not found that means event is not registered (not being served). In this way, no matter the number webhook to be served, just write the function to handle it, in webhook_events.py
import webhook_events def process_webhook(event): event_name = event['name'] function_name = event_name.replace('.', '_') function = getattr(webhook_events, function_name, None) if function: function(event) else: print('Event %s is not registered.' % event_name)
More robust and pythonic way of implementing process_webhook is by using decorators. Lets define a decorator in webhook_events.py which will map the event_name to its function. Here the EVENT_MAP is dictionary inside a setting module, which will contain event name as key and event function as its value.
from django.conf import settings def register(event_name): def wrapper(event_function): # Initializing settings.event_map if not already event_map = getattr(settings, 'EVENT_MAP', None) if not event_map: settings.EVENT_MAP = dict() # Mapping event name to its function settings.EVENT_MAP[event_name] = event_function return event_function return wrapper @register('event.one') def event_one(event): # do something for # for event 'event.one' @register('event.two') def event_two(event): # do something for # for event 'event.two'
In this case, the process_webhook will look like below:
def process_webhook(event): event_name = event['name'] function = settings.EVENT_MAP.get(event_name, None) if function: function(event) else: print('Event %s is not registered.' % event_name)
This is the way which I prefer to implement webhook handler in python. How would you prefer ? Please feel free to comment below.