NOTE: Forked from github_webhooks as I needed to access more things:
- query params from incoming webhooks for identification
background_tasksto handle long running webhook actions.
Simple and lightweight micro framework for quick integration with GitHub
webhooks.
It's based on FastAPI and pydantic, nothing more!
Async and mypy friendly.
Just add github-webhooks-framework2 package.
Example:
pip install github-webhooks-framework2poetry add github-webhooks-framework2
Create file example.py and copy next code:
import uvicorn
from pydantic import BaseModel
from github_webhooks import create_app
from github_webhooks.schemas import WebhookCommonPayload
# WebhookCommonPayload is based on pydantic.BaseModel
class PullRequestPayload(WebhookCommonPayload):
class Pull(BaseModel):
title: str
url: str
action: str
pull_request: Pull
# Initialize Webhook App
app = create_app()
# Register webhook handler:
# `pull_request` - name of an event to handle
# `PullRequestPayload` - webhook payload will be parsed into this model
@app.hooks.register('pull_request', PullRequestPayload)
async def handler(payload: PullRequestPayload) -> None:
print(f'New pull request {payload.pull_request.title}')
print(f' link: {payload.pull_request.url}')
print(f' author: {payload.sender.login}')
if __name__ == '__main__':
# start uvicorn server
uvicorn.run(app)We start by defining payload Model to parse incoming Pull Request Body.
class PullRequestPayload(WebhookCommonPayload):
class Pull(BaseModel):
title: str
url: str
action: str
pull_request: PullIn this example we only want to get action, pull_request.title and pull_request.url from payload.
By subclassing WebhookCommonPayload model will automatically get sender, repository and organization fields.
Next - we are creating ASGI app (based on FastAPI app)
app = create_app()Optionally we can provide here secret_token Github Webhook secret
app = create_app(secret_token='super-secret-token')And time to define our handler
@app.hooks.register('pull_request', PullRequestPayload)
async def handler(payload: PullRequestPayload) -> None:
print(f'New pull request {payload.pull_request.title}')
print(f' link: {payload.pull_request.url}')
print(f' author: {payload.sender.login}')We are using here @app.hooks.register deco, which accepts 2 arguments:
event: str- name of webhook eventpayload_cls: pydantic.BaseModel- pydantic model class to parse request, subclassed frompydantic.BaseModelorWebhookCommonPayload.
And our handler function may include the following (named) params:
async def handler(payload: PullRequestPayload, headers: WebhookHeaders, query_params: QueryParams, background_tasks: BackgroundTasks) -> Optional[str]:
# `headers` will be WebhookHeaders model with Github Webhook headers parsed.
...An example setup can be found in example/server.py.
- Install local venv:
python -m venv .venv - Activate:
. .venv/bin/activate - Install deps with
pip install -r requirements.txt -r requirements-test.txt - When done with packages:
./freeze-requirements.sh