我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用yaml.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 parse_yaml(stream, overrides={}): import yaml import sys from autocmake.interpolate import interpolate try: config = yaml.load(stream, yaml.SafeLoader) except yaml.YAMLError as exc: print(exc) sys.exit(-1) for k in config: if k in overrides: config[k] = overrides[k] config = interpolate(config, config) return config
def ordered_load( stream, Loader=yaml.SafeLoader, object_pairs_hook=OrderedDict): """Load yaml as an ordered dict This allows us to inspect the order of the file on disk to make sure it was correct by our rules. """ 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) # for parameters.yaml we treat numbers (especially version # numbers) as strings. So that microversion specification of 2.20 # and 2.2 don't get confused. OrderedLoader.add_constructor( u'tag:yaml.org,2002:float', yaml.constructor.SafeConstructor.construct_yaml_str) return yaml.load(stream, OrderedLoader)
def importyaml(connection,metadata,sourcePath): eveGraphics = Table('eveGraphics',metadata) print "Importing Graphics" print "opening Yaml" with open(os.path.join(sourcePath,'fsd','graphicIDs.yaml'),'r') as yamlstream: print "importing" trans = connection.begin() graphics=load(yamlstream,Loader=SafeLoader) print "Yaml Processed into memory" for graphic in graphics: connection.execute(eveGraphics.insert(), graphicID=graphic, sofFactionName=graphics[graphic].get('sofFactionName',''), graphicFile=graphics[graphic].get('graphicFile',''), sofHullName=graphics[graphic].get('sofHullName',''), sofRaceName=graphics[graphic].get('sofRaceName',''), description=graphics[graphic].get('description','')) trans.commit()
def importyaml(connection,metadata,sourcePath): print "Importing BSD Tables" files=glob.glob(os.path.join(sourcePath,'bsd','*.yaml')) for file in files: head, tail = os.path.split(file) tablename=tail.split('.')[0] print tablename tablevar = Table(tablename,metadata) print "Importing {}".format(file) print "Opening Yaml" trans = connection.begin() with open(file,'r') as yamlstream: rows=load(yamlstream,Loader=SafeLoader) print "Yaml Processed into memory" for row in rows: connection.execute(tablevar.insert().values(row)) trans.commit()
def parse_yaml(self, data, path_hint=None): ''' convert a yaml string to a data structure. Also supports JSON, ssssssh!!!''' stripped_data = data.lstrip() loaded = None if stripped_data.startswith("{") or stripped_data.startswith("["): # since the line starts with { or [ we can infer this is a JSON document. try: loaded = json.loads(data) except ValueError as ve: if path_hint: self.module.fail_json(msg=path_hint + ": " + str(ve)) else: self.module.fail_json(msg=str(ve)) else: # else this is pretty sure to be a YAML document loaded = yaml.load(data, Loader=Loader) return loaded
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 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)
