Created
March 24, 2012 04:53
-
-
Save tresbailey/2178403 to your computer and use it in GitHub Desktop.
Often times, when parsing a user-inputted JSON converted to dict object for a RESTful web service, I have a need to verify the high-level structure of the JSON. In the REST service, we may have many resources with various needed input dicts. Thus I want
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
TYPE_PARSERS = { | |
'http://www.w3.org/2001/XMLSchema.xsd#string': str, | |
'http://www.w3.org/2001/XMLSchema.xsd#int': int, | |
'http://www.w3.org/2001/XMLSchema.xsd#float': float, | |
'http://www.w3.org/2001/XMLSchema.xsd#date': { 'module': 'datetime', | |
'klass': 'datetime', | |
'method': 'strptime', | |
'args': '%d-%m-%y'}, | |
'http://www.w3.org/2001/XMLSchema.xsd#datetime': { 'module': 'datetime', | |
'klass': 'datetime', | |
'method': 'strptime', | |
'args': '%d-%m-%y %H:%M'}, | |
'http://www.w3.org/2001/XMLSchema.xsd#hexBinary': { 'module': 'uuid', | |
'klass': 'uuid', | |
'method': 'UUID'}, | |
'http://www.w3.org/2001/XMLSchema.xsd#localElement': dict | |
} | |
def convert_value_to_def(func_name, field_value): | |
""" | |
Converts a value from any possible input type to the defined type from above | |
""" | |
if isinstance(func_name, dict): | |
# Setup for more complex conversions, like Dates | |
m_imp = __import__(func_name.get('module'), | |
globals(), locals(), | |
[func_name.get('klass')], -1) | |
try: | |
klass = getattr(m_imp, func_name.get('klass')) | |
except AttributeError: | |
klass = m_imp | |
func = getattr(klass, func_name.get('method')) | |
convert = func(field_value, func_name.get('args')) | |
elif func_name is not None: | |
# For simple conversions, like string | |
convert = func_name(field_value) | |
return convert | |
class JSON_Struct(object): | |
""" | |
Type checking class that behaves like a callable after instantiation | |
""" | |
def __init__(self, sub_types=None, addl_funks=None): | |
""" | |
Create the checking object | |
:arg: sub_types - dict defining the keys to search for in the input dict | |
The keys of sub_types are the keys that are available for the input | |
The values of sub_types are a dict with keys - 'required' and 'types' | |
required - boolean, Is the tag required in the input dict | |
types - list, predefined types from TYPE_PARSERS or a user-defined type | |
:arg: addl_funks - dict defining the conversion callable for the user-defined types | |
The keys of addl_funks are the user-defined types (normally nested dict structs) | |
The values of addl_funks are the conversion callables (normally an instance of JSON_Struct) | |
""" | |
self.sub_types = sub_types or {} | |
self.addl_funks = addl_funks or {} | |
def __call__(self, args): | |
""" | |
Allows the instance to be called as a function to validate the structure of the input dict | |
:arg: args - input dict from the user that should be validated | |
""" | |
missed_flds = [tag for tag, specs in self.sub_types.iteritems() | |
if specs.get('required', False) and tag not in args.keys()] | |
if missed_flds: | |
raise KeyError | |
for key, value in args.iteritems(): | |
if self.sub_types.has_key('_'): | |
key = '_' | |
elif not self.sub_types.has_key(key): | |
raise KeyError | |
avail_types = self.sub_types[key]['types'] | |
raised_errors = list() | |
converted = None | |
for sub_type in avail_types: | |
func_name = TYPE_PARSERS.get(sub_type, | |
self.addl_funks.get(sub_type)) | |
try: | |
converted = convert_value_to_def(func_name, value) | |
except Exception, e: | |
raised_errors.append(e) | |
if converted is None and raised_errors: | |
raise ValueError | |
if __name__ == '__main__': | |
""" | |
Test for handling a basic structure of a 1-level dict | |
""" | |
simple = JSON_Struct({'a': {'types': ['http://www.w3.org/2001/XMLSchema.xsd#string']}, | |
'b': {'types': ['http://www.w3.org/2001/XMLSchema.xsd#int']}}) | |
conv = simple({'a': 'some string', 'b': '100'}) | |
""" | |
Test for handling unvalidated 1-level dict | |
""" | |
try: | |
conv = simple({'a': 'some string', 'b': 'bad string'}) | |
raise Exception | |
except (KeyError, ValueError): | |
pass | |
""" | |
Test for nested 2-level dict | |
""" | |
simple = JSON_Struct({'c': {'types': ['http://www.w3.org/2001/XMLSchema.xsd#string']}, | |
'd': {'types': ['http://www.w3.org/2001/XMLSchema.xsd#string']}}) | |
complex = JSON_Struct({'a': {'types': ['http://www.w3.org/2001/XMLSchema.xsd#int', | |
'http://www.w3.org/2001/XMLSchema.xsd#float']}, | |
'b': {'types': ['simple_def']}}, | |
{'simple_def': simple}) | |
conv = complex({'a': '45', 'b': {'c': 'some string', 'd': 'another'}}) | |
""" | |
Test for recursive nested dict with nested child requirements | |
The 'b' tag accepts another a and b dict, and for the children, the 'a' tag is required | |
""" | |
simple = JSON_Struct({'a': {'types': ['http://www.w3.org/2001/XMLSchema.xsd#string']}}) | |
import copy | |
copied = copy.deepcopy(simple) | |
copied.sub_types['a'].update({'required': True}) | |
copied.sub_types.update({'b': {'types': ['child_req']}}) | |
copied.addl_funks.update({'child_req': copied}) | |
simple.sub_types.update({'b': {'types': ['child_req']}}) | |
simple.addl_funks.update({'child_req': copied}) | |
conv = simple({'a': 'some', 'b': {'a': 'foo'}}) | |
conv = simple({'a': 'foo', 'b': {'a': 'bar', 'b': {'a': 'baz', 'b': {'a': 'bay'}}}}) | |
conv = simple({'b': {'a': 'foobar'}}) |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment