def _get_loaded_template(self):
"""
Converts template document to dict replacing entities UIDs with objects
:return: dict -> loaded YAML document
"""
if hasattr(self, '_loaded_template'):
return self._loaded_template
preprocessed = self._get_preprocessed_template()
class CustomLoader(yaml.SafeLoader):
pass
class TemplateObject(yaml.YAMLObject):
yaml_tag = '!kd'
yaml_loader = CustomLoader
@classmethod
def from_yaml(cls, loader, node):
return self._entities_by_uid[loader.construct_scalar(node)]
patt = re.compile(r'^(?:{0})$'.format('|'.join(self._entities_by_uid)))
CustomLoader.add_implicit_resolver('!kd', patt, None)
try:
self._loaded_template = yaml.load(preprocessed,
Loader=CustomLoader)
except (yaml.scanner.ScannerError, yaml.parser.ParserError):
raise PredefinedAppExc.UnparseableTemplate
return self._loaded_template
python类YAMLObject()的实例源码
def set_serialize_params_recursive(self, obj):
base_arg_names = map(lambda x: x[0], inspect.getmembers(yaml.YAMLObject))
if not isinstance(obj, Serializable):
raise RuntimeError("attempting deserialization of non-Serializable object %s of type %s" % (str(obj), type(obj)))
init_args = self.get_init_args(obj)
class_param_names = [x[0] for x in inspect.getmembers(obj.__class__)]
init_args.remove("self")
obj.serialize_params = {}
items = inspect.getmembers(obj)
for name, val in items:
if name=="yaml_context":
raise ValueError("'yaml_context' is a reserved specifier, please rename argument")
if name in base_arg_names or name.startswith("_") or name in ["serialize_params", "init_params", "kwargs"] or name in class_param_names: continue
if isinstance(val, Serializable):
obj.serialize_params[name] = val
self.set_serialize_params_recursive(val)
elif type(val) in [type(None), bool, int, float, str, type(six.u("")), datetime.datetime, dict, set]:
obj.serialize_params[name] = val
elif type(val)==list:
obj.serialize_params[name] = val
for item in val:
if isinstance(item, Serializable):
self.set_serialize_params_recursive(item)
else:
continue
if not name in init_args:
raise ValueError("unknown init parameter for %s: %s" % (obj.yaml_tag, name))
obj.init_params = dict(obj.serialize_params)
def _register_with_representer(cls):
# TODO: check if we could/should just inherit from yaml.YAMLObject
# or could may be craft our own metaclass
yaml.SafeDumper.add_representer(cls, SpecObject.yaml_representer)
