Source code for nti.webhooks.zcml

# -*- coding: utf-8 -*-
Support for configuring webhook delivery using ZCML.

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

from zope.configuration.fields import GlobalObject
from zope.configuration.fields import Text

from zope.interface import Interface

from import Permission
from zope.schema import TextLine

from nti.webhooks.subscriptions import getGlobalSubscriptionManager
from nti.webhooks.interfaces import IWebhookSubscription
from nti.webhooks._schema import ObjectEventInterface

# pylint:disable=inherit-non-class

[docs]class Path(Text): """ Accepts a single absolute traversable path. Unlike :class:`zope.configuration.fields.Path`, this version requires that the path be absolute and uses URL separators. """
[docs] def fromUnicode(self, value): result = super(Path, self).fromUnicode(value) if not result or not result.startswith('/'): raise ValueError() # pragma: no cover XXX: This should be something specific. return result
[docs]class IStaticSubscriptionDirective(Interface): """ 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_ = GlobalObject( title=IWebhookSubscription['for_'].title, description=IWebhookSubscription['for_'].description, default=IWebhookSubscription['for_'].default, required=False, ) when = ObjectEventInterface( title=IWebhookSubscription['when'].title, description=IWebhookSubscription['when'].description, default=IWebhookSubscription['when'].default, required=False, ) to = IWebhookSubscription['to'].bind(None) dialect = TextLine( # We can't use the field directly because it wants to validate # against a Choice using a vocabulary based on registered utilities. # That doesn't work as an argument if we're still registering those # utilities. title=IWebhookSubscription['dialect_id'].title, description=IWebhookSubscription['dialect_id'].description, default=IWebhookSubscription['dialect_id'].default, required=False ) owner = IWebhookSubscription['owner_id'].bind(None) permission = Permission( title=u"The permission to check", description=u""" 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``. """, required=False )
[docs]class IStaticPersistentSubscriptionDirective(IStaticSubscriptionDirective): """ Define a local, static, persistent subscription. Local persistent subscriptions live in the ZODB database, beneath some :class:``. They are identified by a traversable path beginning from the root of the database; note that this may not be the exact same as a path exposed in the application because this path will need to include the name of the root application object, while application paths typically do not. This package uses :mod:`zope.generations` to keep track of registered subscriptions and synchronize the database with what is in executed ZCML. Thus it is very important not to remove ZCML directives, or only execute part of the ZCML configuration unless you intend for the subscriptions not found in ZCML to be removed. All the options are the same as for :class:`IStaticSubscriptionDirective`, with the addition of the required ``site_path``. """ site_path = Path( title=u'The path to traverse to the site', description=u"A persistent subscription manager will be installed in this site.", required=True, )
# XXX: Active/inactive. Should be able to keep one without losing the history.
[docs]class IDialectDirective(Interface): """ Create a new dialect subclass of `~.DefaultWebhookDialect` and register it as a utility named *name*. """ # REMEMBER: Keep this in sync with the fields defined in # DefaultWebhookDialect. name = TextLine( title=u"Name", description=u"Name of the dialect registration. Limited to ASCII characters.", required=True, ) externalizer_name = TextLine( title=u"The name of the externalization adapters to use", description=u"Remember, if adapters by this name do not exist, the default will be used.", required=False, ) externalizer_policy_name = TextLine( title=u'The name of the externalizer policy component to use.', description=u""" .. important:: An empty string selects the :mod:`nti.externalization` default policy, which uses Unix timestamps. To use the default policy of :mod:`nti.webhooks`, omit this argument. """, required=False, ) http_method = TextLine( # Perhaps this should be a choice. title=u"The HTTP method to use.", description=u"This should be a valid method name, but that's not enforced", required=False, ) user_agent = TextLine( title=u"The User-Agent header string to use.", required=False, )
def _static_subscription_action(subscription_kwargs): getGlobalSubscriptionManager().createSubscription(**subscription_kwargs) def _check_sub_kwargs(kwargs): to = kwargs.pop('to') for_ = kwargs.pop('for_', None) or IStaticSubscriptionDirective['for_'].default when = kwargs.pop('when', None) or IStaticSubscriptionDirective['when'].default owner = kwargs.pop("owner", None) permission = kwargs.pop('permission', None) dialect = kwargs.pop('dialect', None) if kwargs: # pragma: no cover raise TypeError subscription_kwargs = dict(to=to, for_=for_, when=when, owner_id=owner, permission_id=permission, dialect_id=dialect) return subscription_kwargs def static_subscription(context, **kwargs): # type: (zope.configuration.config.ConfigurationMachine, dict) -> None subscription_kwargs = _check_sub_kwargs(kwargs) context.action( # No conflicts. You can register as many identical hooks # as you want. discriminator=None, callable=_static_subscription_action, args=(subscription_kwargs,), # Try to execute towards the end so any validation that needs # previous directives, like permission lookup, can work. order=9999 ) def _persistent_subscription_action(site_path, subscription_kwargs): from .generations import get_schema_manager schema = get_schema_manager() schema.addSubscription(site_path, subscription_kwargs) def persistent_subscription(context, site_path=None, **kwargs): # We need to add the utility and subscription each time we're invoked # in case of z3c.baseregistry, I think. The discriminators should make sure # there's only one. subscription_kwargs = _check_sub_kwargs(kwargs) context.action( discriminator=(site_path, tuple(subscription_kwargs.items())), callable=_persistent_subscription_action, args=(site_path, subscription_kwargs), ) def dialect_directive(context, name=None, **kwargs): # By not specifying the kwargs (giving defaults), anything that's not # in the ZCML will be completely absent. from zope.component.zcml import handler from nti.webhooks.dialect import DefaultWebhookDialect from nti.webhooks.interfaces import IWebhookDialect if 'externalizer_policy_name' in kwargs and not kwargs['externalizer_policy_name']: # This is how you specify to use the nti.externalization default. kwargs['externalizer_policy_name'] = None dialect = type( 'ZCMLWebhookDialect-' + str(name), (DefaultWebhookDialect,), kwargs )() context.action( discriminator=('webhook-dialect', name), callable=handler, args=('registerUtility', dialect, IWebhookDialect, name,, )