Skip to content

Instantly share code, notes, and snippets.

Last active October 13, 2020 11:59
Show Gist options
  • Save evilUrge/af71ae35bad54a23634b6646a581360c to your computer and use it in GitHub Desktop.
Save evilUrge/af71ae35bad54a23634b6646a581360c to your computer and use it in GitHub Desktop.
Plain webhook - Report for every change in a spesific model based on a pre-exists django WebHook model condition
import logging
logger = logging.getLogger(__name__)
def report(saved_object):
report is the main function in reporter
# :param uid: django ORM unique identifier; for internal usage.
:param saved_object: the object that being saved, MUST HAVE A ORGANISATION ATTACHED TO IT!
:return: Builds a req body, make a request, logs it in LOG model and return Boolean for success
import requests
from itertools import chain
from datetime import datetime
from django.apps import apps
from services.reporter.models import WebHook, Log
# We need to import all models here, because the service is dynamic - it
# can accept any model (so not just BillingRequest):
globals().update({_class.__name__: _class for _class in apps.get_models()})
def _execute():
def _build_req_body(body_obj, saved_object=saved_object):
Builds a JSON body for a request (in case needed and as long as it's not a GET req)
Fetch model values when a value of one of the body_obj is in the following format: `{DjangoServiceModel.field}`
:param body_obj: example of a unprocessed obj (template for the 'body' field of the Webhook model):
"message": "Your static message goes here!",
"timestamp": "this will be auto-populated",
"{BillingRequest.global_id}": "this will be auto-populated"
"body_message": "This will show the following EXTERNAL obj id:
" {{DemoModel.object.get(user=self).id}}"
#! use the key "self" to use the triggered obj
#! for filtering in an external model.
:return: all timestamps getting signed with current timestamp, special template(SEE COMMENT ABOVE FOR EXAMPLE)
and returns a new modified body.
example of a processed obj (what will be outputted by the service):
"message": "Your static message goes here!",
"timestamp": "2019-10-31T13:15:40",
"djangoObjId": "QmlsbGluZ1JlcXVlc3ROb2RlOjQ2"
def _find_model_value(term, flatten=False):
if term.find('{') is not -1 and term.find('}') is not -1:
terms = term[term.find('{') + 1:term.find('}')].split('.')
if terms.__len__() > 1:
model_name, fields_name = terms[0], terms[1:]
if model_name and fields_name and globals().get(model_name):
model = globals().get(model_name)
first_field = fields_name[0] if type(fields_name) in (list, tuple) else fields_name
if first_field == 'objects' and type(fields_name) in (list, tuple):
for field in fields_name:
if all(field.find(term) is not -1 for term in ['(', ')', 'self']):
final_attrs = list(filter(None, term[term.find(')') + 1:-1].split('.')))
attrs = list(
filter(None, [field[:field.find('(')], field[field.find(')') + 1:]]))
metdata = {}
for key, value in {k: v for k, v in [y.split('=') for y in \
field.find('(') + 1:field.find(')')].split(
metdata.update({key: saved_object if value == "self" else value})
return ' '.join(list(chain(*[[
getattr(ex_model, final_attr) for final_attr in final_attrs] for
ex_model \
in [getattr(getattr(globals().get('obj', model), first_field), attr) \
(**metdata) for attr in attrs]])))
except AttributeError as err:
return ''
obj = getattr(globals().get('obj', model), field)
except AttributeError as err:
return obj.__str__() if flatten else {terms[-1]: obj} if globals().get('obj') else {}
elif hasattr(model, first_field):
obj = getattr(model.objects.get(id=uid), first_field, '')
if type(fields_name) in (list, tuple):
for field in fields_name[1:]:
obj = getattr(obj, field) if hasattr(obj, field) else obj
return obj.__str__() if flatten else {terms[-1]: obj}
else:'Failed to find fields:{".".join(fields_name)} in model:{model_name}')
else:"Couldn't find any relevant fields")
return '' if flatten else {}
result = {}
for key, value in body_obj.items():
if type(value) is dict:
raw_value = {key: _build_req_body(value)}
if key.find('{') is not -1 and key.find('}') is not -1:
raw_value = _find_model_value(key)
elif key == 'timestamp':
raw_value = {'timestamp':'%Y-%m-%dT%H:%M:%S')}
if value.find('{') is not -1 and value.find('}') is not -1:
rebuild_value = []
_find_model_value(word, True) if word.find('{') is not -1 and word.find(
'}') is not -1 else
(type(word) in (list, tuple, dict) and ' '.join(
type(word) is dict and word.values() or word)) or word)
for word in value.split()]
value = ' '.join(rebuild_value)
raw_value = {key: value}
return result
body = {} if org_obj.request_type == 'get' else {'json': _build_req_body(org_obj.body)}
if body:
response = getattr(requests, org_obj.request_type)(org_obj.url, **body)
Log.objects.create(webhook=org_obj, passed=response.status_code == 200,
request=body.get('json', ''),
response=response.content if response.status_code == 200 else response.reason)
if response.status_code == 200:
return True
logger.error('{} - {}'.format(response.status_code, response.reason))
except (requests.exceptions.HTTPError, requests.exceptions.ConnectionError) as err:
Log.objects.create(webhook=org_obj, passed=False, request=body.get('json', ''), response='FAILED')
logger.error('{} - {}'.format('FAILED', err.args[0].pool))
return False
organisation, uid = False, False
for org_type in ['type1', 'type2', 'type3']:
if getattr(saved_object, org_type, False) \
or hasattr(saved_object, 'organisation') and getattr(saved_object.organisation, org_type, False):
from services.organisations.models import Organisation
organisation, uid = Organisation.objects.get(**{
org_type: getattr(saved_object, org_type) if hasattr(saved_object, org_type) else getattr(
saved_object.organisation, org_type, False)}),
except Organisation.DoesNotExist:"Failed to find the required organisation")
if not organisation and uid:"Failed to find the required organisation")
if not WebHook.objects.filter(organisation=organisation).exists():"No webhook available")
for org_obj in WebHook.objects.filter(organisation=organisation):
Check if there's any conditions for execution
ex: {field: expected_value, field2: expected_value2,field3: [expected_value,expected_value2,expected_value3]}
if org_obj.conditions.items().__len__() is 0:
for field, expected_value in org_obj.conditions.items():
if hasattr(saved_object, field):
if type(expected_value) in (str, bool, int, float):
getattr(saved_object, field) == expected_value and _execute()
elif type(expected_value) in (list, tuple):
getattr(saved_object, field) in expected_value and _execute()
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment