Typed interactions with Github Webhooks
Project description
Octohook
Octohook makes working with incoming Github Webhooks extremely easy.
It parses the incoming payload into Python classes to allow for auto-complete and other goodies. For example, octohook
provides functions for payload values which require string interpolation.
For example, almost every repository payload has an archive_url
with some required and conditional parameters.
{
"repository" : {
"archive_url": "https://api.github.com/repos/doodla/octohook-playground/{archive_format}{/ref}"
}
}
The Repository
model provides an archive_url()
method which has archive_format
as an argument and ref
as an optional variable.
>>> repo.archive_url("hello")
https://api.github.com/repos/doodla/octohook-playground/hello
>>> repo.archive_url("hello","world")
https://api.github.com/repos/doodla/octohook-playground/hello/world"
Gotchas
Github doesn't send consistent payloads for each model necessitating that the non-optional model type hints conform to the least common denominator.
Depending on the event type, you can get more information for a particular model, or less.
For example, Github sends a changes
key with some payloads with the edited
action. For other actions, the key is not present. In such cases, our event.changes
is None
.
This can happen for arbitrary payloads, so I'd suggest tailoring your code to the expected incoming webhook.
If anyone has a good suggestion on how to tackle this issue, feel free to email me, or create a PR!
Because Github sends different payloads for a combination of event type
and action
, unless I have access to all the variations, I cannot be sure that the corresponding model is correct.
Current coverage is documented here. If you can provide any of the missing events, please make a PR.
Sample Usage
app.py
from flask import Flask, request, Response
import octohook
from octohook.events import PullRequestEvent
app = Flask(__name__)
@app.route('/webhook', methods=['POST'])
def webhook():
github_event = request.headers.get('X-GitHub-Event') # pull_request
# Assuming the incoming event was a PullRequestEvent
event : PullRequestEvent = octohook.parse(github_event, request.json)
# Do work with this event
return Response(event.pull_request.head.user.name, status=200)
@hook
Alternatively, you can also let octohook
do the heavy lifting of finding and executing the appropriate handlers for any given webhook.
The @hook
decorator takes in four parameters, the WebhookEvent
, a list of WebhookEventAction
s, an optional list of repositories and a debug
flag (defaults to False
).
Any function this decorator is applied to is invoked whenever you receive an event with the specified WebhookEvent
and a listed WebhookEventAction
.
If present, the hook can be filtered for one or more repositories in a multi-repository set up.
Note: The "full_name" of the repositories are used. Ex: "doodla/octohook"
If you set debug=True
on any @hook
, only those hooks fire for the corresponding webhook event.
@hook(WebhookEvent.PULL_REQUEST,[WebhookEventAction.CREATED, WebhookEventAction.EDITED])
def work(event: PullRequestEvent):
pass
work()
is automatically called with the parsed PullRequestEvent
anytime you receive a webhook event with X-Github-Event: pull_request
and it has any of the created
or edited
actions.
If you don't specify a list of actions, then the function is invoked for any action. For some events like Push
, which do not have an action
, take care not to specify any actions in the decorator.
hooks/do_something.py
from octohook import hook,WebhookEvent,WebhookEventAction
from octohook.events import LabelEvent,PullRequestEvent
@hook(WebhookEvent.LABEL, [WebhookEventAction.CREATED])
def runs_when_label_event_with_created_action_is_received(event: LabelEvent):
print(event.label.name)
@hook(WebhookEvent.PULL_REQUEST)
def runs_when_pull_request_event_with_any_action_is_received(event: PullRequestEvent):
print(event.changes)
app.py
from flask import Flask, request, Response
import octohook
app = Flask(__name__)
octohook.load_hooks(["hooks"])
@app.route('/webhook', methods=['POST'])
def webhook():
github_event = request.headers.get('X-GitHub-Event')
octohook.handle_webhook(event_name=github_event, payload=request.json)
return Response("OK", status=200)
handle_hooks
goes through all the handlers sequentially and blocks till everything is done. Any exceptions are logged to logging.getLogger('octohook')
. You can configure the output stream of this logger to capture the logs.
Model Overrides
octohook
provides a way to extend/modify the models being provided in the event object. model_overrides
is a dictionary where you can map octohook
models to your own.
import octohook
from octohook.models import PullRequest
class MyPullRequest(PullRequest):
def custom_work(self):
pass
octohook.load_hooks(["module_a"])
octohook.model_overrides = {
PullRequest: MyPullRequest
}
Now, everytime octohook
attempts to initialize a PullRequest
object, it will initialize MyPullRequest
instead.
Check the test for example usage.
Note
- The class is initialized with the relevant
payload: dict
data from the incoming event payload. - It is recommended you subclass the original model class, but it is not required.
- Type hints are no longer reliable for the overridden classes.
Project details
Release history Release notifications | RSS feed
Download files
Download the file for your platform. If you're not sure which to choose, learn more about installing packages.
Source Distribution
Built Distribution
Hashes for octohook-0.12.0-py3-none-any.whl
Algorithm | Hash digest | |
---|---|---|
SHA256 | 4776c4fbcdb63a858e8198f935034ef9c80d875e86225013717732ac0cbc8fe4 |
|
MD5 | 4f8ac91e8a6f4a921c1a0e5725ce655b |
|
BLAKE2b-256 | 18b5366d0ead85b2e629d35f93a24375ea3c1174a9b0cbe759f046770aa409b2 |