Start refactoring of the trigger system:
- 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
This commit is contained in:
parent
2ff77df946
commit
49b575afb6
25 changed files with 2449 additions and 1602 deletions
79
util/dict_wrappers.py
Normal file
79
util/dict_wrappers.py
Normal file
|
@ -0,0 +1,79 @@
|
|||
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
|
Reference in a new issue