Customizing HTTP Requests¶
Once an active subscription matches and is applicable for a certain combination of object and event, eventually it’s time to create the actual HTTP request (body and headers) that will be delivered to the target URL.
This document will outline how that is done, and discuss how to customize that process. It will use static subscriptions to demonstrate, but these techniques are equally relevant for dynamic subscriptions.
Let’s begin by registering an example static subscription, and
refresh our memory of what the HTTP request looks like by default.
First, the imports and creation of the static subscription; we’ll use
the objects defined in employees.py
:
from persistent import Persistent
from zope.container.contained import Contained
from zope.site.folder import Folder
from zope.site import LocalSiteManager
from zope.annotation.interfaces import IAttributeAnnotatable
from zope.interface import implementer
class Employees(Folder):
def __init__(self):
Folder.__init__(self)
self['employees'] = Folder()
self.setSiteManager(LocalSiteManager(self))
class Department(Employees):
pass
class Office(Employees):
pass
@implementer(IAttributeAnnotatable)
class Employee(Contained, Persistent):
COUNTER = 0
def __init__(self):
self.__counter__ = self.COUNTER
Employee.COUNTER += 1
def __repr__(self):
return "<Employee %s %d>" % (
self.__name__,
self.__counter__,
)
class ExternalizableEmployee(Employee):
def toExternalObject(self, **kwargs):
return self.__name__
from zope.testing import cleanup
cleanup.addCleanUp(lambda: setattr(Employee, 'COUNTER', 0))
>>> import transaction
>>> from zope import lifecycleevent, component
>>> from zope.container.folder import Folder
>>> from employees import Employee
>>> from zope.configuration import xmlconfig
>>> from nti.webhooks.interfaces import IWebhookSubscriptionManager
>>> from nti.webhooks.interfaces import IWebhookDeliveryManager
>>> 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://example.com/some/path"
... for="employees.Employee"
... when="zope.lifecycleevent.interfaces.IObjectCreatedEvent" />
... </configure>
... """)
>>> from nti.webhooks.testing import mock_delivery_to
>>> mock_delivery_to('https://example.com/some/path')
Next, we trigger the subscription and wait for it to be delivered.
>>> def trigger_delivery(factory=Employee, name=u'Bob', last_modified=None):
... _ = transaction.begin()
... employee = factory()
... employee.__name__ = name
... if last_modified: employee.LastModified = last_modified
... lifecycleevent.created(employee)
... transaction.commit()
... component.getUtility(IWebhookDeliveryManager).waitForPendingDeliveries()
>>> trigger_delivery()
Finally, we can look at what we actually sent. It’s not too pretty.
>>> sub_manager = component.getUtility(IWebhookSubscriptionManager)
>>> subscription = sub_manager['Subscription']
>>> attempt = subscription.pop()
>>> attempt.response.status_code
200
>>> print(attempt.request.url)
https://example.com/some/path
>>> print(attempt.request.method)
POST
>>> import pprint
>>> pprint.pprint({str(k): str(v) for k, v in attempt.request.headers.items()})
{'Accept': '*/*',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
'Content-Length': '84',
'Content-Type': 'application/json',
'User-Agent': 'nti.webhooks...'}
>>> print(attempt.request.body)
{"Class": "NonExternalizableObject", "InternalType": "<class 'employees.Employee'>"}
Customizing The Body¶
There are a few different ways to customize the body, and they can be applied at the same time.
The first way to customize the body is to register an adapter
producing an IWebhookPayload
. The
adapter can be an adapter for just the object, or it can be a
multi-adapter from the object and the event that triggered the
subscription. By default, both an adapter named
DefaultWebhookDialect.externalizer_name
and the unnamed
adapter are attempted. When such an adapter is found, its value is
externalized instead of the target of the event.
Note
While a single adapter is frequently enough, multi-adapters are allowed in case the context of the event matters. For example, one might wish to externalize something different when an object is created versus when it is modified or deleted.
Important
The security checks described in Delivery Security Checks apply to the object of the triggering event, not the adapted value.
Single Adapters¶
Working from lowest priority to highest priority, let’s demonstrate some adapters.
First, an adapter for a single object with no name.
>>> from zope.interface import implementer
>>> from zope.component import adapter
>>> from nti.webhooks.interfaces import IWebhookPayload
>>> @implementer(IWebhookPayload)
... @adapter(Employee)
... def single_adapter(employee):
... return employee.__name__
>>> component.provideAdapter(single_adapter)
Triggering the event now produces a different body.
>>> trigger_delivery()
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
"Bob"
>>> trigger_delivery(name=u'Susan')
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
"Susan"
Higher priority is a named adapter.
>>> from zope.component import named
>>> @implementer(IWebhookPayload)
... @adapter(Employee)
... @named("webhook-delivery")
... def named_single_adapter(folder):
... return "An Employee"
>>> component.provideAdapter(named_single_adapter)
>>> trigger_delivery()
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
"An Employee"
Of course, if the object already provides IWebhookPayload
,
then it is returned directly without using those adapters.
>>> @implementer(IWebhookPayload)
... class PayloadFactory(Employee):
... """An employee that is its own payload."""
>>> trigger_delivery(factory=PayloadFactory)
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
{"Class": "NonExternalizableObject", "InternalType": "<class 'PayloadFactory'>"}
Multi-adapters¶
Multi-adapters are the highest priority. They take precedence over the object itself
being a IWebhookPayload
already.
The unnamed adapter for the event and the object is higher priority than the named single adapter or the object itself.
>>> from zope.lifecycleevent.interfaces import IObjectCreatedEvent
>>> @implementer(IWebhookPayload)
... @adapter(Employee, IObjectCreatedEvent)
... def multi_adapter(employee, event):
... return "employee-and-event"
>>> component.provideAdapter(multi_adapter)
>>> trigger_delivery()
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
"employee-and-event"
Finally, the highest priority is a named multi-adapter.
>>> from zope.lifecycleevent.interfaces import IObjectCreatedEvent
>>> @implementer(IWebhookPayload)
... @adapter(Employee, IObjectCreatedEvent)
... @named("webhook-delivery")
... def named_multi_adapter(employee, event):
... return "named-employee-and-event"
>>> component.provideAdapter(named_multi_adapter)
>>> trigger_delivery()
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
"named-employee-and-event"
Cleanup¶
Let’s remove all those adapters and get back to a base state.
>>> gsm = component.getGlobalSiteManager()
>>> gsm.unregisterAdapter(named_multi_adapter, name=named_multi_adapter.__component_name__)
True
>>> gsm.unregisterAdapter(multi_adapter)
True
>>> gsm.unregisterAdapter(named_single_adapter, name=named_single_adapter.__component_name__)
True
>>> gsm.unregisterAdapter(single_adapter)
True
Webhook Dialects¶
Another way to customize the body, and much more, is to write a
dialect. Every subscription is associated, by name, with a
dialect. Dialects are registered utilities that implement
nti.webhooks.interfaces.IWebhookDialect
; there is a global
default (the empty name, ‘’) dialect implemented in
DefaultWebhookDialect
. When defining new dialects, you
should extend this class. In fact, the behaviour defined above is
implemented by this class in its
DefaultWebhookDialect.produce_payload()
method.
Important
Dialects must not be persistent objects. They may be used outside of contexts where ZODB is available.
Note
Much of what is done next with custom code can also be done with ZCML. See Defining A Dialect Using ZCML for details.
Setting the Body¶
One easy way to customize the body is to use named externalizers. The
default dialect uses an externalizer with the name given in
externalizer_name
; a subclass can
change this by setting it on the class object. We’ll demonstrate by
first defining and registering a
IInternalObjectExternalizer
with a custom name.
>>> from nti.externalization.interfaces import IInternalObjectExternalizer
>>> from nti.externalization import to_standard_external_dictionary
>>> @implementer(IInternalObjectExternalizer)
... @adapter(Employee)
... @named('webhook-testing')
... class EmployeeExternalizer(object):
... def __init__(self, context):
... self.context = context
... def toExternalObject(self, **kwargs):
... std = to_standard_external_dictionary(self.context, **kwargs)
... std['Class'] = 'Employee'
... std['Name'] = self.context.__name__
... return std
>>> component.provideAdapter(EmployeeExternalizer)
Next, we’ll create a dialect that uses this externalizer, and register it:
>>> from nti.webhooks.dialect import DefaultWebhookDialect
>>> @named('webhook-testing')
... class TestDialect(DefaultWebhookDialect):
... externalizer_name = 'webhook-testing'
>>> component.provideUtility(TestDialect())
We then alter the subscription to use this dialect:
>>> subscription.dialect_id = 'webhook-testing'
Now when we trigger the subscription, we use this externalizer:
>>> trigger_delivery()
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
{"Class": "Employee", "Name": "Bob"}
Customizing the Body With Externalization Policies
Making smaller tweaks can be accomplished by adjusting the
externalization policy that’s used. By default, the externalization
policy, named in
externalizer_policy_name
, produces
ISO8601 strings for values stored as Unix timestamps (seconds since
the epoch).
>>> trigger_delivery(last_modified=123456789.0)
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
{"Class": "Employee", "Last Modified": "1973-11-29T21:33:09Z", "Name": "Bob"}
The best way to adjust this is to set the externalizer_policy_name
to a different value. A unicode string should refere to a registered
externalization policy component. If we set it to None
, the
default policy (which outputs the timestamps as numbers) is used.
>>> TestDialect.externalizer_policy_name = None
>>> trigger_delivery(last_modified=123456789.0)
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
{"Class": "Employee", "Last Modified": 123456789.0, "Name": "Bob"}
Setting Headers¶
The dialect is also responsible for customizing aspects of the HTTP request, including headers, authentication, and the method. Our previous attempt used the default values for these things:
>>> print(attempt.request.method)
POST
>>> import pprint
>>> pprint.pprint({str(k): str(v) for k, v in attempt.request.headers.items()})
{'Accept': '*/*',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
'Content-Length': '66',
'Content-Type': 'application/json',
'User-Agent': 'nti.webhooks ...'}
Lets apply some simple customizations and send again.
>>> TestDialect.http_method = 'PUT'
>>> TestDialect.user_agent = 'doctests'
>>> trigger_delivery()
>>> attempt = subscription.pop()
>>> print(attempt.request.method)
PUT
>>> pprint.pprint({str(k): str(v) for k, v in attempt.request.headers.items()})
{'Accept': '*/*',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
'Content-Length': '36',
'Content-Type': 'application/json',
'User-Agent': 'doctests'}
Defining A Dialect Using ZCML¶
For the simple cases that are customizations of the strings defined by
the DefaultWebhookDialect
, you can use a ZCML directive to define them.
-
interface
nti.webhooks.zcml.
IDialectDirective
[source] Create a new dialect subclass of
DefaultWebhookDialect
and register it as a utility named name.-
name
¶ Name
Name of the dialect registration. Limited to ASCII characters.
Implementation: zope.schema.TextLine
Read Only: False Required: True Default Value: None Allowed Type: str
-
externalizer_name
¶ The name of the externalization adapters to use
Remember, if adapters by this name do not exist, the default will be used.
Implementation: zope.schema.TextLine
Read Only: False Required: False Default Value: None Allowed Type: str
-
externalizer_policy_name
¶ The name of the externalizer policy component to use.
Important
An empty string selects the
nti.externalization
default policy, which uses Unix timestamps. To use the default policy ofnti.webhooks
, omit this argument.Implementation: zope.schema.TextLine
Read Only: False Required: False Default Value: None Allowed Type: str
-
We can repeat the above example using just ZCML.
>>> from nti.webhooks.subscriptions import resetGlobals
>>> resetGlobals()
>>> 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:webhookDialect
... name='zcml-dialect'
... externalizer_name='webhook-testing'
... externalizer_policy_name=''
... http_method='PUT'
... user_agent='zcml-tests' />
... <webhooks:staticSubscription
... dialect='zcml-dialect'
... to="https://example.com/some/path"
... for="employees.Employee"
... when="zope.lifecycleevent.interfaces.IObjectCreatedEvent" />
... </configure>
... """)
>>> subscription = sub_manager['Subscription']
>>> trigger_delivery(last_modified=123456789.0)
>>> attempt = subscription.pop()
>>> print(attempt.request.body)
{"Class": "Employee", "Last Modified": 123456789.0, "Name": "Bob"}
>>> print(attempt.request.method)
PUT
>>> pprint.pprint({str(k): str(v) for k, v in attempt.request.headers.items()})
{'Accept': '*/*',
'Accept-Encoding': 'gzip, deflate',
'Connection': 'keep-alive',
'Content-Length': '66',
'Content-Type': 'application/json',
'User-Agent': 'zcml-tests'}