Source code for nti.webhooks.dialect

# -*- coding: utf-8 -*-
"""
Implementations of dialects.

"""
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function

import pkg_resources
from zope.interface import implementer
from zope import component

from requests import Request

from nti import externalization
from nti.externalization.interfaces import IExternalObjectRepresenter
from nti.webhooks.interfaces import IWebhookDialect
from nti.webhooks.interfaces import IWebhookPayload

[docs]@implementer(IWebhookDialect) class DefaultWebhookDialect(object): """ Default implementation of a :class:`nti.webhooks.interfaces.IWebhookDialect`. This class is intended to be subclassed; other dialect implementations *should* extend this class. This permits freedom in adding additional methods to the interface. """ # REMEMBER: Keep the ZCML directive in sync with # this as much as possible. #: The name of the externalizer used to produce the #: external form. This is also the highest-priority name #: of the adapter used. externalizer_name = u'webhook-delivery' #: The name of the externalization policy utility #: used to produce the external form. This defaults to one #: that uses ISO8601 format for Unix timestamps. externalizer_policy_name = u'webhook-delivery' #: Which representation to use. Passed to #: :func:`nti.externaliaztion.to_external_representation` externalizer_format = externalization.representation.EXT_REPR_JSON #: The MIME type of the body produced by :meth:`externalizeData`. #: If you change the :attr:`externalizer_format`, you need to change #: this value. content_type = 'application/json' #: The HTTP "User-Agent" header. user_agent = 'nti.webhooks %s' % ( pkg_resources.require('nti.webhooks')[0].version, ) #: The HTTP method (verb) to use. http_method = 'POST'
[docs] def produce_payload(self, data, event): """ produce_payload(data, event) -> IWebhookPayload Non-interface method. Given data delivered through an event, try to find a ``IWebhookPayload`` for it. From highest to lowest priority, this means: - A multi-adapter from the object and the event named :attr:`externalizer_name`. - The unnamed multi-adapter. - A single adapter from the object named :attr:`externalizer_name` - The unnamed single adapter. The *data* is used as the context for the lookup in all cases. XXX: Or maybe we should use the subscription as the context? Note that if there exists an adapter registration that returns None, we continue with lower-priority adapters. """ result = component.queryMultiAdapter((data, event), IWebhookPayload, name=self.externalizer_name, context=data) if result is not None: return result result = component.queryMultiAdapter((data, event), IWebhookPayload, context=data) if result is not None: return result if IWebhookPayload.providedBy(data): return data result = component.queryAdapter(data, IWebhookPayload, name=self.externalizer_name, context=data) if result is not None: return result return component.queryAdapter(data, IWebhookPayload, default=data, context=data)
[docs] def externalizeData(self, data, event): "See :meth:`nti.webhooks.interfaces.IWebhookDialect.externalizeData`" payload = self.produce_payload(data, event) # ext_data = externalization.to_external_representation( # payload, # ext_format=self.externalizer_format, # name=self.externalizer_name) # ``to_external_representation`` doesn't accept the policy or policy_name # argument in nti.externalization 2.0 policy_name = self.externalizer_policy_name if policy_name is None: # We do this here to keep clients from knowing the # ugly details. from nti.externalization._base_interfaces import NotGiven policy_name = NotGiven ext = externalization.to_external_object( payload, name=self.externalizer_name, policy_name=policy_name) rep = component.getUtility(IExternalObjectRepresenter, name=self.externalizer_format) return rep.dump(ext)
def produce_headers(self, http_session, subscription, attempt): headers = { 'Content-Type': self.content_type, 'User-Agent': self.user_agent } return headers
[docs] def prepareRequest(self, http_session, subscription, attempt): "See :meth:`nti.webhooks.interfaces.IWebhookDialect.prepareRequest`" headers = self.produce_headers(http_session, subscription, attempt) http_request = Request(self.http_method, subscription.to, headers=headers, data=attempt.payload_data) return http_session.prepare_request(http_request)