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'}})