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.

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:

urls.py

1from django.conf.urls import url
2import views
3
4urlpatterns = [
5    url(r'^webhook', views.webhook, name='webhook'),
6]

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 let's load the webhook data and send 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_exempt decorator above the view function. Also put an @require_post decorator to ensure the request is only POST.

views.py

1from django.views.decorators.csrf import csrf_exempt
2from django.views.decorators.http import require_POST
3import json
4
5@require_POST
6@csrf_exempt
7def webhook(request):
8    # Load the event data from JSON
9    data = json.loads(request.body)
10    
11    # And process it
12    process_webhook(data)
13    
14    return HttpResponse('Processed.', status=200)

The above implementation of URL endpoint will remain different for various other Python web frameworks like Flask, Tornado, Twisted. But the process_webhook function implementation below will remain same irrespective of any framework.

Processing Event

There may be different type of events we need to handle. So, before proceeding to implement process_webhook function, let's 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.

webhook_events.py

1def event_one(event):
2    # do something for event 'event.one'
3    pass
4
5def event_two(event):
6    # do something for event 'event.two'
7    pass

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 implementations based on extendability. Most basic version is below:

Basic Implementation

1import webhook_events
2
3def process_webhook(event):
4    event_name = event['name']
5    
6    if event_name == 'event.one':
7        webhook_events.event_one(event)
8    elif event_name == 'event.two':
9        webhook_events.event_two(event)
10    # and so on

A Better Way

Now suppose, there are 10s of webhooks 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.

1import webhook_events
2
3def process_webhook(event):
4    event_name = event['name']
5    function_name = event_name.replace('.', '_')
6    
7    function = getattr(webhook_events, function_name, None)
8    
9    if function:
10        function(event)
11    else:
12        print('Event %s is not registered.' % event_name)

Using Decorators

More robust and Pythonic way of implementing process_webhook is by using decorators. Let's define a decorator in webhook_events.py which will map the event_name to its function.

webhook_events.py with Decorators

1from django.conf import settings
2
3def register(event_name):
4    def wrapper(event_function):
5        # Initializing settings.EVENT_MAP if not already
6        event_map = getattr(settings, 'EVENT_MAP', None)
7        if not event_map:
8            settings.EVENT_MAP = dict()
9        
10        # Mapping event name to its function
11        settings.EVENT_MAP[event_name] = event_function
12        
13        return event_function
14    return wrapper
15
16@register('event.one')
17def event_one(event):
18    # do something for event 'event.one'
19    pass
20
21@register('event.two')
22def event_two(event):
23    # do something for event 'event.two'
24    pass

In this case, the process_webhook will look like below:

Final process_webhook Implementation

1def process_webhook(event):
2    event_name = event['name']
3    
4    function = settings.EVENT_MAP.get(event_name, None)
5    
6    if function:
7        function(event)
8    else:
9        print('Event %s is not registered.' % event_name)

This is the way which I prefer to implement webhook handler in Python. The decorator pattern provides a clean, scalable solution that makes it easy to add new webhook handlers without modifying the core processing logic.