Object-level Web Hook application.
This project was heavily influence by django-rest-hooks.
pip install django-objects-hooks
setup your settings:
INSTALLED_APPS = (
...
'doh', # mandatory
)
HOOK_ALLOWED_MODELS = [
'yourapp.Vehicle' # example
]
create web hooks:
vehicle = Vehicle.objects.get(type='bus', name='55')
Hook.objects.create(
user=user, content_object=vehicle,
target='http://www.example.com/vehicles',
)
and that's it!
Whenever a vehicle
is saved, DOH handles the post_save
signal and initiates the hook delivering process.
class Vehicle(models.Model):
...
def get_static_payload(self):
# if this method exists, the payload will be fetched only once
# and used for every hook within an event for this model
return model_to_dict(self)
def get_dynamic_payload(self, hook):
# if this method exists, the payload will be fetched for each
# hook in any event for this model
return {
'resource_uri': 'http://foo.com/vehicles/{pk}'.format(pk=self.pk)
'action': hook.action,
}
By default the payload is an empty json ({}
).
from doh.signals import hook_event
hook_event.send(sender=Vehicle, instance=bus55, action='crashed', payload={})
HOOK_COLLECTION_DELIVERER
: responsible for filtering theHook
model and dumping each payload.HOOK_ELEMENT_DELIVERER
: responsible for sending a POST request to a target with a given payload.
HOOK_COLLECTION_DELIVERER = "doh.deliverers.deliver_hooks_using_task"
HOOK_ELEMENT_DELIVERER = "doh.deliverers.deliver_hook_using_task"
You can just override one setting thus combining with the default deliverers.
HOOK_COLLECTION_DELIVERER = "doh.deliverers.deliver_hooks_using_gevent"
or
HOOK_COLLECTION_DELIVERER = "doh.deliverers.deliver_hooks_using_gevent_task"
HOOK_ELEMENT_DELIVERER
is not used in this case.
Create your own custom deliverers from HooksDeliverer
and HookDeliverer
:
from doh.deliverers.base import HooksDeliverer, HookDeliverer
class CustomHooksDeliverer(HooksDeliverer):
def dump_payload(self, payload):
# you may override this method to use simplejson instead of ujson
return simplejson.dumps(payload)
def after_deliver(self, hooks):
# you may use this method to do some logging
logger.info('foobar')
deliver_hooks = CustomHooksDeliverer.deliver
deliver_hooks_using_task = CustomHooksDeliverer.delay
class CustomHookDeliverer(HookDeliverer):
def after_deliver(self, response):
# you may use this method to delete hooks
if response.status_code in [404, 410]:
Hook.object.filter(target=response.url).delete()
deliver_hook = CustomHookDeliverer.deliver
deliver_hook_using_task = CustomHookDeliverer.delay