def Deserializer(stream_or_string, **options):
"""
Deserialize a stream or string of YAML data.
"""
if isinstance(stream_or_string, bytes):
stream_or_string = stream_or_string.decode('utf-8')
if isinstance(stream_or_string, six.string_types):
stream = StringIO(stream_or_string)
else:
stream = stream_or_string
try:
for obj in PythonDeserializer(yaml.load(stream, Loader=SafeLoader), **options):
yield obj
except GeneratorExit:
raise
except Exception as e:
# Map to deserializer error
six.reraise(DeserializationError, DeserializationError(e), sys.exc_info()[2])
python类SafeLoader()的实例源码
def Deserializer(stream_or_string, **options):
"""
Deserialize a stream or string of YAML data.
"""
if isinstance(stream_or_string, bytes):
stream_or_string = stream_or_string.decode('utf-8')
if isinstance(stream_or_string, six.string_types):
stream = StringIO(stream_or_string)
else:
stream = stream_or_string
try:
for obj in PythonDeserializer(yaml.load(stream, Loader=SafeLoader), **options):
yield obj
except GeneratorExit:
raise
except Exception as e:
# Map to deserializer error
six.reraise(DeserializationError, DeserializationError(e), sys.exc_info()[2])
def __init__(self, config=None):
"""Instantiate configuration."""
self.raw_config = {}
self.config_path = config
# These get assigned in style_configuration decorator.
self._style_task = None
self._style_flow = None
self._style_condition = None
self._style_condition_foreach = None
self._style_storage = None
self._style_edge = None
self._style_store_edge = None
self._style_graph = None
self._style_fallback_edge = None
self._style_fallback_node = None
self._style_fallback_true = None
if self.config_path is not None:
try:
with open(self.config_path) as input_file:
self.raw_config = yaml.load(input_file, Loader=yaml.SafeLoader)
except Exception as exc:
raise RequestError("Unable to load or parse style configuration file %r: %s"
% (self.config_path, str(exc))) from exc
def __init__(self, config=None):
"""Instantiate configuration."""
self.raw_config = {}
self.config_path = config
# These get assigned in style_configuration decorator.
self._style_task = None
self._style_flow = None
self._style_condition = None
self._style_condition_foreach = None
self._style_storage = None
self._style_edge = None
self._style_store_edge = None
self._style_graph = None
self._style_fallback_edge = None
self._style_fallback_node = None
self._style_fallback_true = None
if self.config_path is not None:
try:
with open(self.config_path) as input_file:
self.raw_config = yaml.load(input_file, Loader=yaml.SafeLoader)
except Exception as exc:
raise RequestError("Unable to load or parse style configuration file %r: %s"
% (self.config_path, str(exc))) from exc
def Deserializer(stream_or_string, **options):
"""
Deserialize a stream or string of YAML data.
"""
if isinstance(stream_or_string, bytes):
stream_or_string = stream_or_string.decode('utf-8')
if isinstance(stream_or_string, six.string_types):
stream = StringIO(stream_or_string)
else:
stream = stream_or_string
try:
for obj in PythonDeserializer(yaml.load(stream, Loader=SafeLoader), **options):
yield obj
except GeneratorExit:
raise
except Exception as e:
# Map to deserializer error
six.reraise(DeserializationError, DeserializationError(e), sys.exc_info()[2])
def ordered_load(stream, Loader=yaml.SafeLoader, object_pairs_hook=OrderedDict):
"""Load a yaml configuration into an OrderedDict."""
class OrderedLoader(Loader):
pass
def construct_mapping(loader, node):
loader.flatten_mapping(node)
return object_pairs_hook(loader.construct_pairs(node))
OrderedLoader.add_constructor(yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG, construct_mapping)
return yaml.load(stream, OrderedLoader)
def from_yaml(cls, data):
return cls.from_dict(
load(data, Loader=SafeLoader),
)
def yaml_load(stream):
"""Load YAML document, but load all strings as unicode on py2."""
import yaml
class UnicodeLoader(yaml.SafeLoader):
"""Yaml SafeLoader Class, default encoding is UTF-8."""
pass
# NOTE:
# In [2]: yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG
# Out[2]: 'tag:yaml.org,2002:map'
UnicodeLoader.add_constructor(
yaml.resolver.BaseResolver.DEFAULT_SCALAR_TAG,
UnicodeLoader.construct_scalar)
return yaml.load(stream, UnicodeLoader)
def load(stream, Loader=yaml.SafeLoader, object_pairs_hook=OrderedDict):
class OrderedLoader(Loader):
pass
def construct_mapping(loader, node):
loader.flatten_mapping(node)
return object_pairs_hook(loader.construct_pairs(node))
OrderedLoader.add_constructor(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
construct_mapping)
return yaml.load(stream, OrderedLoader)
def save(data, stream=None, Dumper=yaml.SafeDumper,
default_flow_style=False,
encoding='utf-8',
**kwds):
class OrderedDumper(Dumper):
pass
def _dict_representer(dumper, data):
return dumper.represent_mapping(
yaml.resolver.BaseResolver.DEFAULT_MAPPING_TAG,
data.items())
OrderedDumper.add_representer(OrderedDict, _dict_representer)
OrderedDumper.add_representer(np.float64,
lambda dumper, data: dumper.represent_float(float(data)))
OrderedDumper.add_representer(complex,
lambda dumper, data: dumper.represent_str(str(data)))
OrderedDumper.add_representer(np.complex128,
lambda dumper, data: dumper.represent_str(str(data)))
OrderedDumper.add_representer(np.ndarray,
lambda dumper, data: dumper.represent_list(list(data)))
# I added the following two lines to make pyrpl compatible with pyinstruments. In principle they can be erased
if isinstance(data, dict) and not isinstance(data, OrderedDict):
data = OrderedDict(data)
return yaml.dump(data,
stream=stream,
Dumper=OrderedDumper,
default_flow_style=default_flow_style,
encoding=encoding,
**kwds)
# usage example:
# load(stream, yaml.SafeLoader)
# save(data, stream=f, Dumper=yaml.SafeDumper)
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
def run_c7n_mailer(logger):
email_config_filepath = '/custodian/email/email-config.yml'
email_config = yaml.load(get_file_contents(email_config_filepath), Loader=yaml.SafeLoader)
session = boto3.Session()
mailer_sqs_queue_processor = MailerSqsQueueProcessor(email_config, session, logger)
mailer_sqs_queue_processor.run(parallel=False)
def __init__(self, *args, **kwargs):
yaml.SafeLoader.__init__(self, *args, **kwargs)
self.add_constructor(u'tag:yaml.org,2002:map', type(self).construct_yaml_map)
self.add_constructor(u'tag:yaml.org,2002:omap', type(self).construct_yaml_map)
def importyaml(connection,metadata,sourcePath):
eveIcons = Table('eveIcons',metadata)
print "Importing Icons"
print "Opening Yaml"
with open(os.path.join(sourcePath,'fsd','iconIDs.yaml'),'r') as yamlstream:
trans = connection.begin()
icons=load(yamlstream,Loader=SafeLoader)
print "Yaml Processed into memory"
for icon in icons:
connection.execute(eveIcons.insert(),
iconID=icon,
iconFile=icons[icon].get('iconFile',''),
description=icons[icon].get('description',''))
trans.commit()
def importyaml(connection,metadata,sourcePath):
print "Importing Categories"
invCategories = Table('invCategories',metadata)
trnTranslations = Table('trnTranslations',metadata)
print "opening Yaml"
trans = connection.begin()
with open(os.path.join(sourcePath,'fsd','categoryIDs.yaml'),'r') as yamlstream:
print "importing"
categoryids=load(yamlstream,Loader=SafeLoader)
print "Yaml Processed into memory"
for categoryid in categoryids:
connection.execute(invCategories.insert(),
categoryID=categoryid,
categoryName=categoryids[categoryid].get('name',{}).get('en','').decode('utf-8'),
iconID=categoryids[categoryid].get('iconID'),
published=categoryids[categoryid].get('published',0))
if (categoryids[categoryid].has_key('name')):
for lang in categoryids[categoryid]['name']:
try:
connection.execute(trnTranslations.insert(),tcID=6,keyID=categoryid,languageID=lang,text=categoryids[categoryid]['name'][lang].decode('utf-8'));
except:
print '{} {} has a category problem'.format(categoryid,lang)
trans.commit()
def importyaml(connection,metadata,sourcePath):
certCerts = Table('certCerts',metadata)
certSkills = Table('certSkills',metadata,)
skillmap={"basic":0,"standard":1,"improved":2,"advanced":3,"elite":4}
print "Importing Certificates"
print "opening Yaml"
with open(os.path.join(sourcePath,'fsd','certificates.yaml'),'r') as yamlstream:
trans = connection.begin()
certificates=load(yamlstream,Loader=SafeLoader)
print "Yaml Processed into memory"
for certificate in certificates:
connection.execute(certCerts.insert(),
certID=certificate,
name=certificates[certificate].get('name',''),
description=certificates[certificate].get('description',''),
groupID=certificates[certificate].get('groupID'))
for skill in certificates[certificate]['skillTypes']:
for skillLevel in certificates[certificate]['skillTypes'][skill]:
connection.execute(certSkills.insert(),
certID=certificate,
skillID=skill,
certLevelInt=skillmap[skillLevel],
certLevelText=skillLevel,
skillLevel=certificates[certificate]['skillTypes'][skill][skillLevel]
)
trans.commit()
def test_yaml_load():
ystr = yaml.dump({'a' : 1, 'b' : 2, 'c' : 3})
y = yaml.load(ystr)
yaml.dump(y)
y = yaml.load(ystr, Loader=yaml.SafeLoader)
def parse(tmpl_str):
"""Takes a string and returns a dict containing the parsed structure.
This includes determination of whether the string is using the
JSON or YAML format.
"""
# strip any whitespace before the check
tmpl_str = tmpl_str.strip()
if tmpl_str.startswith('{'):
tpl = json.loads(tmpl_str)
else:
try:
tpl = yaml.load(tmpl_str, Loader=yaml_loader)
except yaml.YAMLError:
# NOTE(prazumovsky): we need to return more informative error for
# user, so use SafeLoader, which return error message with template
# snippet where error has been occurred.
try:
tpl = yaml.load(tmpl_str, Loader=yaml.SafeLoader)
except yaml.YAMLError as yea:
raise ValueError(yea)
else:
if tpl is None:
tpl = {}
# Looking for supported version keys in the loaded template
if not ('HeatTemplateFormatVersion' in tpl
or 'heat_template_version' in tpl
or 'AWSTemplateFormatVersion' in tpl):
raise ValueError(_("Template format version not found."))
return tpl
def parse(env_str):
"""Takes a string and returns a dict containing the parsed structure.
This includes determination of whether the string is using the
YAML format.
"""
try:
env = yaml.load(env_str, Loader=template_format.yaml_loader)
except yaml.YAMLError:
# NOTE(prazumovsky): we need to return more informative error for
# user, so use SafeLoader, which return error message with template
# snippet where error has been occurred.
try:
env = yaml.load(env_str, Loader=yaml.SafeLoader)
except yaml.YAMLError as yea:
raise ValueError(yea)
else:
if env is None:
env = {}
elif not isinstance(env, dict):
raise ValueError(_('The environment is not a valid '
'YAML mapping data type.'))
for param in env:
if param not in SECTIONS:
raise ValueError(_('environment has wrong section "%s"') % param)
return env
def yaml_load(value):
if yaml is None:
raise RuntimeError("Yaml not available")
return yaml.load(value, Loader=SafeLoader)
