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 of nti.webhooks, omit this argument.

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

The HTTP method to use.

This should be a valid method name, but that’s not enforced

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

The User-Agent header string to use.

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'}