49b575afb6
- Move each trigger handler into its own file - Add dictionary helper classes for easier reading and writing of dict-based data - Extract the web hook payload -> internal representation building for each trigger system - Add tests for this transformation - Remove support for Github archived-based building
79 lines
2.3 KiB
Python
79 lines
2.3 KiB
Python
import json
|
|
from jsonpath_rw import parse
|
|
|
|
class SafeDictSetter(object):
|
|
""" Specialized write-only dictionary wrapper class that allows for setting
|
|
nested keys via a path syntax.
|
|
|
|
Example:
|
|
sds = SafeDictSetter()
|
|
sds['foo.bar.baz'] = 'hello' # Sets 'foo' = {'bar': {'baz': 'hello'}}
|
|
sds['somekey'] = None # Does not set the key since the value is None
|
|
"""
|
|
def __init__(self, initial_object=None):
|
|
self._object = initial_object or {}
|
|
|
|
def __setitem__(self, path, value):
|
|
self.set(path, value)
|
|
|
|
def set(self, path, value, allow_none=False):
|
|
""" Sets the value of the given path to the given value. """
|
|
if value is None and not allow_none:
|
|
return
|
|
|
|
pieces = path.split('.')
|
|
current = self._object
|
|
|
|
for piece in pieces[:len(pieces)-1]:
|
|
current_obj = current.get(piece, {})
|
|
if not isinstance(current_obj, dict):
|
|
raise Exception('Key %s is a non-object value: %s' % (piece, current_obj))
|
|
|
|
current[piece] = current_obj
|
|
current = current_obj
|
|
|
|
current[pieces[-1]] = value
|
|
|
|
def dict_value(self):
|
|
""" Returns the dict value built. """
|
|
return self._object
|
|
|
|
def json_value(self):
|
|
""" Returns the JSON string value of the dictionary built. """
|
|
return json.dumps(self._object)
|
|
|
|
|
|
class JSONPathDict(object):
|
|
""" Specialized read-only dictionary wrapper class that uses the jsonpath_rw library
|
|
to access keys via an X-Path-like syntax.
|
|
|
|
Example:
|
|
pd = JSONPathDict({'hello': {'hi': 'there'}})
|
|
pd['hello.hi'] # Returns 'there'
|
|
"""
|
|
def __init__(self, dict_value):
|
|
""" Init the helper with the JSON object.
|
|
"""
|
|
self._object = dict_value
|
|
|
|
def __getitem__(self, path):
|
|
def raise_exception():
|
|
raise KeyError('Unknown path: %s' % path)
|
|
|
|
return self.get(path, not_found_handler=raise_exception)
|
|
|
|
def get(self, path, not_found_handler=None):
|
|
""" Returns the value found at the given path. Path is a json-path expression. """
|
|
jsonpath_expr = parse(path)
|
|
matches = jsonpath_expr.find(self._object)
|
|
if not matches:
|
|
return not_found_handler() if not_found_handler else None
|
|
|
|
match = matches[0].value
|
|
if not match:
|
|
return not_found_handler() if not_found_handler else None
|
|
|
|
if isinstance(match, dict):
|
|
return JSONPathDict(match)
|
|
|
|
return match
|