API Docs

Invenio module for processing webhook events.

class invenio_webhooks.ext.InvenioWebhooks(app=None, **kwargs)

Invenio-Webhooks extension.

Extension initialization.

init_app(app, entry_point_group='invenio_webhooks.receivers')

Flask application initialization.

init_config(app)

Initialize configuration.

Models for webhook receivers.

class invenio_webhooks.models.CeleryReceiver(receiver_id)

Asynchronous receiver.

Receiver which will fire a celery task to handle payload instead of running it synchronously during the request.

Initialize a receiver identifier.

class invenio_webhooks.models.Event(**kwargs)

Incoming webhook event data.

Represents webhook event data which consists of a payload and a user id.

A simple constructor that allows initialization from kwargs.

Sets attributes on the constructed instance using the names and values in kwargs.

Only keys that are present as attributes of the instance’s class are allowed. These could be, for example, any mapped columns or relationships.

classmethod create(receiver_id, user_id=None)

Create an event instance.

id

Event identifier.

payload

Store payload in JSON format.

payload_headers

Store payload headers in JSON format.

process()

Process current event.

receiver

Return registered receiver.

receiver_id

Receiver identifier.

response

Store response in JSON format.

response_headers

Store response headers in JSON format.

user_id

User identifier.

validate_receiver(key, value)

Validate receiver identifier.

exception invenio_webhooks.models.InvalidPayload

Raised when the payload is invalid.

exception invenio_webhooks.models.InvalidSignature

Raised when the signature does not match.

class invenio_webhooks.models.Receiver(receiver_id)

Base class for a webhook receiver.

A receiver is responsible for receiving and extracting a payload from a request. You must implement run method that accepts the event instance.

Initialize a receiver identifier.

check_signature()

Check signature of signed request.

extract_payload()

Extract payload from request.

get_hook_url(access_token)

Get URL for webhook.

In debug and testing mode the hook URL can be overwritten using WEBHOOKS_DEBUG_RECEIVER_URLS configuration variable to allow testing webhooks via services such as e.g. Ultrahook.

WEBHOOKS_DEBUG_RECEIVER_URLS = dict(
    github='http://github.userid.ultrahook.com',
)

run(event)

Implement method accepting the Event instance.

signature = ''

Default signature.

exception invenio_webhooks.models.ReceiverDoesNotExist

Raised when receiver does not exist.

exception invenio_webhooks.models.WebhookError

General webhook error.

Configuration

Webhooks module.

invenio_webhooks.config.WEBHOOKS_DEBUG_RECEIVER_URLS = {}

Mapping of receiver id to URL pattern.

This allows generating URLs to an intermediate webhook proxy service like Ultrahook for testing on development machines:

WEBHOOKS_DEBUG_RECEIVER_URLS = {
    'github': 'https://hook.user.ultrahook.com/?access_token=%%(token)s'
}

Signatures

Calculate signatures for payloads.

invenio_webhooks.signatures.check_x_hub_signature(signature, message)

Check X-Hub-Signature used by GitHub to sign requests.

Parameters:

• signature – HMAC signature extracted from request.
• message – Request message.

invenio_webhooks.signatures.get_hmac(message)

Calculate HMAC value of message using WEBHOOKS_SECRET_KEY.

Parameters:message – String to calculate HMAC for.

REST API

Invenio module for processing webhook events.

class invenio_webhooks.views.ReceiverEventListResource

Receiver event hook.

options(receiver_id=None)

Handle OPTIONS request.

post(*args, **kwargs)

Handle POST request.

invenio_webhooks.views.error_handler(f)

Decorator to handle exceptions.

invenio_webhooks.views.view(*args, **kwargs)

Receiver event hook.

Proxies

Helper proxy to the state object.