')
try:
unflattened_payload = unflatten_keys(payload)
r += htmltext('
')
r += format_payload(unflattened_payload)
r += htmltext('
')
except UnflattenKeysException as e:
r += htmltext('
') % (
_('Unable to preview payload.'),
_('Following error occured: '),
e,
)
r += htmltext('
')
return r.getvalue()
def _q_traverse(self, path):
get_request().is_json_marker = True
return super()._q_traverse(path)
def reverse_geocoding(request, *args, **kwargs):
if not ('lat' in request.GET and 'lon' in request.GET):
return HttpResponseBadRequest()
lat = request.GET['lat']
lon = request.GET['lon']
return HttpResponse(misc.get_reverse_geocoding_data(lat, lon), content_type='application/json')
def geocoding(request, *args, **kwargs):
if 'q' not in request.GET:
return HttpResponseBadRequest()
q = request.GET['q']
url = get_publisher().get_geocoding_service_url()
if '?' in url:
url += '&'
else:
url += '?'
url += 'format=json&q=%s' % urllib.parse.quote(q.encode('utf-8'))
url += '&accept-language=%s' % (get_publisher().get_site_language() or 'en')
return HttpResponse(misc.urlopen(url).read(), content_type='application/json')
def validate_condition(request, *args, **kwargs):
condition = {}
condition['type'] = request.GET.get('type') or ''
condition['value'] = request.GET.get('value_' + condition['type']) or ''
hint = {'msg': ''}
try:
Condition(condition).validate()
except ValidationError as e:
hint['msg'] = str(e)
else:
if request.GET.get('warn-on-datetime') == 'true' and condition['type'] == 'django':
variables = re.compile(r'\b(today|now)\b')
filters = re.compile(r'\|age_in_(years|months|days|hours)')
if variables.search(condition['value']) or filters.search(condition['value']):
hint['msg'] = _(
'Warning: conditions are only evaluated when entering the action, '
'you may need to set a timeout if you want it to be evaluated regularly.'
)
return JsonResponse(hint)
class ProvisionAfterJob(AfterJob):
def __init__(self, json_data, **kwargs):
super().__init__(**kwargs)
self.json_data = json_data
def execute(self):
CmdHoboNotify().process_notification(self.json_data)
def provisionning(request):
if not is_url_signed():
raise AccessForbiddenError()
sync = request.GET.get('sync') == '1'
if sync:
CmdHoboNotify().process_notification(get_request().json)
else:
job = ProvisionAfterJob(json_data=get_request().json)
job.run(spool=True)
return JsonResponse({'err': 0})