Implementing Webhook Handler in Python.

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.

Webhook Handler

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.

Processing event

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)

Decorators

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.

4 thoughts on “Implementing Webhook Handler in Python.

  1. Warning: Attempt to read property "entry" on string in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 190 Warning: Trying to access array offset on value of type null in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 190 Warning: Attempt to read property "displayName" on null in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 191
    vikram says:

    well explained

    1. Warning: Attempt to read property "entry" on string in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 190 Warning: Trying to access array offset on value of type null in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 190 Warning: Attempt to read property "displayName" on null in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 191
      Sumeet P says:

      Thanks Vikram.

  2. Warning: Attempt to read property "entry" on string in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 190 Warning: Trying to access array offset on value of type null in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 190 Warning: Attempt to read property "displayName" on null in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 191
    Ryan says:

    Thanks for this. Great Explanation. However, shouldn’t line 13 in the decorator webhook events be:
    `settings.EVENT_MAP[event_name] = event_function`

    1. Warning: Attempt to read property "entry" on string in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 190 Warning: Trying to access array offset on value of type null in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 190 Warning: Attempt to read property "displayName" on null in /var/app/current/wp-content/plugins/jetpack/modules/gravatar-hovercards.php on line 191
      Sumeet P says:

      Yes, you are right.
      Thank you for pointing out that.

Leave a Reply

Your email address will not be published. Required fields are marked *