Configured Global, Transient Webhook Subscriptions

The simplest type of webhook subscription is one that is configured statically, typically at application startup time, and stores no persistent history (with the facilities provided by this package; applications may store their own history, perhaps by listening for delivery events).

This is useful for a number of scenarios, including:

• Development;
• Integration testing;
• Fire-and-forget delivery of frequent events;
• Simple applications.

This package provides ZCML directives to facilitate this. The directives can either be used globally, creating subscriptions that are valid across the entire application.

interface nti.webhooks.zcml.IStaticSubscriptionDirective

Define a global, static, transient subscription.

Static subscriptions are not persistent and live only in the memory of individual processes. Thus, failed deliveries cannot be re-attempted after process shutdown. And of course the delivery history is also transient and local to a process.

for_

The type of object to attempt delivery for.

When object events of type when are fired for instances providing this interface, webhook delivery to target might be attempted.

The default is objects that implement IWebhookPayload.

This is interpreted as for zope.component.registerAdapter() and may name an interface or a type.

Implementation:zope.configuration.fields.GlobalObject Read Only:False Required:False Default Value:<InterfaceClass nti.webhooks.interfaces.IWebhookPayload>

when

The type of event that should result in attempted deliveries.

A type of IObjectEvent, usually one defined in zope.lifecycleevent.interfaces such as IObjectCreatedEvent. The object field of this event must provide the for_ interface; it’s the data from the object field of this event that will be sent to the webhook.

If not specified, all object events involving the for_ interface will be sent.

This must be an interface.

Implementation:nti.webhooks._schema.ObjectEventInterface Read Only:False Required:False Default Value:<InterfaceClass zope.interface.interfaces.IObjectEvent>

Value Type

Implementation:nti.webhooks._schema.ObjectEventField Read Only:False Required:True Default Value:None

to

The complete destination URL to which the data should be sent

This is an arbitrary HTTPS URL. Only HTTPS is supported for delivery of webhooks.

Implementation:nti.webhooks._schema.HTTPSURL Read Only:False Required:True Default Value:None Allowed Type:str

dialect

The ID of the IWebhookDialect to use

Dialects are named utilities. They control the authentication, headers, and HTTP method.

Implementation:zope.schema.TextLine Read Only:False Required:False Default Value:None Allowed Type:str

owner

The ID of the IPrincipal that owns this subscription.

This will be validated at runtime when an event arrives. If the current zope.security.interfaces.IAuthentication utility cannot find a principal with the given ID, the delivery will be failed.

Leave unset to disable security checks.

This cannot be changed after creation.

Implementation:nti.webhooks._schema.PermissivePrincipalId Read Only:False Required:False Default Value:None Allowed Type:str

permission

The permission to check

If given, and an owner is also specified, then only data that has this permission for the owner will result in an attempted delivery. If not given, but an owner is given, this will default to the standard view permission ID, zope.View.

Implementation:zope.security.zcml.Permission Read Only:False Required:False Default Value:None Allowed Type:str

Let’s look at an example of how to use this directive from ZCML. We need to define the XML namespace it’s in, and we need to include the configuration file that defines it. We also need to have the event dispatching provided by zope.component properly set up, as well as some other things described in Configuration. Including this package’s configuration handles all of that.

>>> from zope.configuration import xmlconfig
>>> conf_context = xmlconfig.string("""
... <configure
...     xmlns="http://namespaces.zope.org/zope"
...     xmlns:webhooks="http://nextthought.com/ntp/webhooks"
...     >
...   <include package="nti.webhooks" />
...   <include package="nti.webhooks" file="subscribers_promiscuous.zcml" />
... </configure>
... """, execute=False)

Once that’s done, we can use the webhooks:staticSubscription XML tag to define a subscription to start receiving our webhook deliveries.

ZCML Directive Arguments

There is only one required argument: the destination URL.

The destination must be HTTPS.

>>> conf_context = xmlconfig.string("""
... <configure
...     xmlns="http://namespaces.zope.org/zope"
...     xmlns:webhooks="http://nextthought.com/ntp/webhooks"
...     >
...   <webhooks:staticSubscription to="http://example.com" />
... </configure>
... """, conf_context)
Traceback (most recent call last):
...
zope.configuration.exceptions.ConfigurationError: Invalid value for 'to'
    File "<string>", line 6.2-6.57
    zope.schema.interfaces.InvalidURI: http://example.com

If we specify a permission to check, it must exist.

>>> conf_context = xmlconfig.string("""
... <configure
...     xmlns="http://namespaces.zope.org/zope"
...     xmlns:webhooks="http://nextthought.com/ntp/webhooks"
...     >
...   <webhooks:staticSubscription
...             to="https://example.com"
...             permission="no.such.permission" />
... </configure>
... """, conf_context)
Traceback (most recent call last):
...
zope.configuration.config.ConfigurationExecutionError: File "<string>", line 6.2-8.46
  Could not read source.
    ValueError: ('Undefined permission ID', 'no.such.permission')

Specifying Which Objects

The above (unsuccessful) registration would have tried to send all IObjectEvent events for all objects that implement IWebhookPayload to https://example.com using the default dialect. That’s unlikely to be what you want, outside of tests. Instead, you’ll want to limit the event to particular kinds of objects, and particular events in their lifecycle. The for and when attributes let you do that. Here, we’ll give a comple example saying that whenever a new IContainer is created, we’d like to deliver a webhook.

>>> conf_context = xmlconfig.string("""
... <configure
...     xmlns="http://namespaces.zope.org/zope"
...     xmlns:webhooks="http://nextthought.com/ntp/webhooks"
...     >
...   <include package="zope.component" />
...   <include package="zope.container" />
...   <include package="nti.webhooks" />
...   <include package="nti.webhooks" file="subscribers_promiscuous.zcml" />
...   <webhooks:staticSubscription
...             to="https://this_domain_does_not_exist"
...             for="zope.container.interfaces.IContentContainer"
...             when="zope.lifecycleevent.interfaces.IObjectCreatedEvent" />
... </configure>
... """)

Note

Sometimes, the for and when attributes may not quite get you what you want. You can use an adapter to nti.webhooks.interfaces.IWebhookPayload to derive the desired data. For more, see Customizing HTTP Requests.

Active Subscriptions

Now that we have that in place, let’s verify that it exists as part of the global default IWebhookSubscriptionManager.

>>> from nti.webhooks.interfaces import IWebhookSubscriptionManager
>>> from zope import component
>>> sub_manager = component.getUtility(IWebhookSubscriptionManager)
>>> from zope.interface import verify
>>> verify.verifyObject(IWebhookSubscriptionManager, sub_manager)
True
>>> len(list(sub_manager))
1
>>> name, subscription = list(sub_manager.items())[0]
>>> print(name)
Subscription
>>> subscription
<...Subscription ... to='https://this_domain_does_not_exist' for=IContentContainer when=IObjectCreatedEvent>

And we’ll verify that it is active, by looking for it using the event we just declared:

>>> from nti.webhooks.subscribers import find_active_subscriptions_for
>>> from zope.container.folder import Folder
>>> from zope.lifecycleevent import ObjectCreatedEvent
>>> event = ObjectCreatedEvent(Folder())
>>> active_subscriptions = list(find_active_subscriptions_for(event.object, event))
>>> len(active_subscriptions)
1
>>> active_subscriptions[0]
<...Subscription ... to='https://this_domain_does_not_exist' for=IContentContainer when=IObjectCreatedEvent>
>>> active_subscriptions[0] is subscription
True
>>> subscription.active
True

Next, we need to know if the subscription is applicable to the data. Since we didn’t specify a permission or a principal to check, the subscription is applicable:

See also

Delivery Security Checks for information on security checks.

>>> subscriptions = find_active_subscriptions_for(event.object, event)
>>> [subscription.isApplicable(event.object) for subscription in subscriptions]
[True]

Delivery Attempts

All attempts at delivering a webhook are recorded. Delivery always occurs as a result of committing a transaction, and the resulting attempt object is stored in the corresponding subscription object.

Here, we will briefly look at what happens when we attempt to deliver this webhook. Recall that it uses a domain that does not exist.

See also

Delivery History for more on delivery attempts.

See also

Customizing HTTP Requests for information on customizing what is sent in the delivery attempt.

Unsuccessful Delivery Attempts

Because delivery is transactional, to begin we must be in a transaction:

>>> import transaction
>>> tx = transaction.begin()

Fire the event:

>>> from zope import lifecycleevent
>>> lifecycleevent.created(Folder())

We can see that we have attached a data manager to the transaction:

>>> tx._resources
[<nti.webhooks.datamanager.WebhookDataManager...>]

Don’t Fail The Transaction

However, recall that we specified an invalid domain name, so there is nowhere to attempt to deliver the webhook too. For static webhooks, this is generally a deployment configuration problem and should be attended to by correcting the ZCML. For dynamic subscriptions, the error would be corrected by updating the subscription. This doesn’t fail the commit:

>>> transaction.commit()

But it does record a failed attempt in the subscription:

>>> subscription = sub_manager['Subscription']
>>> len(subscription)
1
>>> attempt = list(subscription.values())[0]
>>> attempt.status
'failed'
>>> print(attempt.message)
Verification of the destination URL failed. Please check the domain.
>>> len(attempt.internal_info.exception_history)
1
>>> print(attempt.internal_info.exception_history[0])
Traceback (most recent call last):
...

Inactive Subscriptions

Subscriptions can be deactivated (made inactive) by asking the manager to do this. The subscription manager is always the subscription’s parent, and deactivating the subscription more than once does nothing.

>>> subscription.__parent__ is sub_manager
True
>>> sub_manager.deactivateSubscription(subscription)
True
>>> sub_manager.deactivateSubscription(subscription)
False
>>> subscription.active
False

Note that we cannot change this attribute directly, it must be done through the manager.

>>> subscription.active = True
Traceback (most recent call last):
...
ValueError:...field is readonly

Inactive subscriptions will not be used for future deliveries, but their existing history is preserved.

>>> len(subscription)
1
>>> find_active_subscriptions_for(event.object, event)
[]
>>> tx = transaction.begin()
>>> lifecycleevent.created(Folder())
>>> tx._resources
[]
>>> transaction.commit()
>>> len(subscription)
1

Of course, inactive subscriptions can be activated again.

>>> sub_manager.activateSubscription(subscription)
True
>>> subscription.active
True
>>> tx = transaction.begin()
>>> lifecycleevent.created(Folder())
>>> transaction.commit()
>>> len(subscription)
2

Removing a subscription from its subscription manager automatically deactivates it.

>>> del sub_manager[subscription.__name__]
>>> subscription.__parent__ is None
True
>>> subscription.active
False