本文整理汇总了Python中yaml.safe_load_all函数的典型用法代码示例。如果您正苦于以下问题:Python safe_load_all函数的具体用法?Python safe_load_all怎么用?Python safe_load_all使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了safe_load_all函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: main
def main():
# First, build a dict of the ruleset definition
ruleset = {}
with open(args.aclTemplate, 'r') as infile:
for entry in yaml.safe_load_all(infile):
ruleset.update(entry)
aclType = ruleset['acl_type']
# Build a crazy dict of dicts of lists to create the appropriate YAML heirachy
yamlRuleset = {}
yamlRuleset['ACL'] = {}
yamlRuleset['ACL']['acl_id'] = ruleset['acl_id']
yamlRuleset['ACL']['acl_type'] = aclType
yamlRuleset['ACL']['description'] = ruleset['description']
yamlRuleset['ACL']['name'] = ruleset['name']
yamlRuleset['ACL']['rules'] = []
# Build a dict of all the rules we know about
rules = {}
for filename in os.listdir(args.rules):
with open(args.rules + filename) as infile:
for entry in yaml.safe_load_all(infile):
rules.update(entry)
# Build a dict of all service definitions
services = {}
for filename in os.listdir(args.service):
with open(args.service + filename, 'r') as infile:
allServices = list(yaml.safe_load_all(infile)) # Convert generator to list
for entry in allServices:
#print("Adding: %s" % entry)
services[entry['name']] = entry
# Build a dict of all the objects we know about
objects = {}
for filename in os.listdir(args.objects):
with open(args.objects + filename) as infile:
for entry in yaml.safe_load_all(infile):
objects.update(entry)
for service in ruleset['SACP']:
service = service.strip("{} ")
if services[service]['acl_type'] != aclType:
print("Error: SACP definition %s is not of type %s" % (services[service]['name'], aclType))
exit(1)
yamlRuleset = parseService(service, services, rules, yamlRuleset)
# Sort ruleset by IP/prefix size
yamlRuleset = addLengths(yamlRuleset, objects)
# Sort the compiled rules list first by dest prefix, then source, finally weight.
yamlRuleset['ACL']['rules'].sort(key=lambda x: (x['_dlen'], x['_slen'], x['_weight']), reverse=True)
print(yaml.dump(yamlRuleset, explicit_start=True, default_flow_style=False))
print('...')
开发者ID:lfintor,项目名称:yaml-netconf,代码行数:57,代码来源:generateACL.py
示例2: get_bridge_reachability_reports
def get_bridge_reachability_reports(s3):
"""
Fetches Tor bridge reachability reports from AWS S3
:param s3:
:param debug:
:return:
"""
reports = []
keys = get_bridge_reachability_report_keys(s3=s3)
for i, key in enumerate(keys):
print("[%d/%d] %s" % (i+1, len(keys), key))
gzipped_yml = s3.get_key(key).get_contents_as_string()
yml = list(yaml.safe_load_all(zlib.decompress(bytes(gzipped_yml), 15+32)))
# Parse YAML file into header/payload, and associated payload with header
header = yml[0]
report = header
report['results'] = []
for subreport in yml[1:]:
if subreport['record_type'] == 'entry':
print("subreport")
report['results'].append(subreport)
if subreport['record_type'] == 'footer':
report['footer'] = subreport
reports.append(report)
return reports
开发者ID:guyhughes,项目名称:ooni-looking-glass,代码行数:29,代码来源:fetch_public_ooni_data.py
示例3: read_config
def read_config(ctx):
filename = os.path.join(os.environ['HOME'], '.teuthology.yaml')
ctx.teuthology_config = {}
with file(filename) as f:
g = yaml.safe_load_all(f)
for new in g:
ctx.teuthology_config.update(new)
开发者ID:tv42,项目名称:teuthology,代码行数:7,代码来源:misc.py
示例4: ArtifactsFromYaml
def ArtifactsFromYaml(yaml_content):
"""Get a list of Artifacts from json."""
try:
raw_list = list(yaml.safe_load_all(yaml_content))
except ValueError as e:
raise ArtifactDefinitionError("Invalid json for artifact: %s" % e)
# Try to do the right thing with json/yaml formatted as a list.
if (isinstance(raw_list, list) and len(raw_list) == 1 and
isinstance(raw_list[0], list)):
raw_list = raw_list[0]
# Convert json into artifact and validate.
valid_artifacts = []
for artifact_dict in raw_list:
# In this case we are feeding parameters directly from potentially
# untrusted yaml/json to our RDFValue class. However, safe_load ensures
# these are all primitive types as long as there is no other deserialization
# involved, and we are passing these into protobuf primitive types.
try:
artifact_value = rdfvalue.Artifact(**artifact_dict)
valid_artifacts.append(artifact_value)
except (TypeError, AttributeError) as e:
raise ArtifactDefinitionError("Invalid artifact definition for %s: %s" %
(artifact_dict.get("name"), e))
return valid_artifacts
开发者ID:ForensicTools,项目名称:GRREAT-475_2141-Chaigon-Failey-Siebert,代码行数:27,代码来源:artifact_lib.py
示例5: find_playbooks
def find_playbooks():
''' find Ansible playbooks'''
all_playbooks = set()
included_playbooks = set()
exclude_dirs = ('adhoc', 'tasks')
for yaml_file in find_files(
os.path.join(os.getcwd(), 'playbooks'),
exclude_dirs, None, r'\.ya?ml$'):
with open(yaml_file, 'r') as contents:
for task in yaml.safe_load_all(contents) or {}:
if not isinstance(task, dict):
# Skip yaml files which are not a dictionary of tasks
continue
if 'include' in task or 'import_playbook' in task:
# Add the playbook and capture included playbooks
all_playbooks.add(yaml_file)
if 'include' in task:
directive = task['include']
else:
directive = task['import_playbook']
included_file_name = directive.split()[0]
included_file = os.path.normpath(
os.path.join(os.path.dirname(yaml_file),
included_file_name))
included_playbooks.add(included_file)
elif 'hosts' in task:
all_playbooks.add(yaml_file)
return all_playbooks, included_playbooks
开发者ID:ingvagabund,项目名称:openshift-ansible,代码行数:29,代码来源:setup.py
示例6: importFromYaml
def importFromYaml():
p = GangaEventProcessor()
count = 0
filename = raw_input('Enter YAML file path: ')
print 'Importing data from "%s"' % filename
try:
f = open(filename, 'rb')
except IOError:
print 'Error! File %s doesn\'t exists!' % filename
return
try:
data=yaml.safe_load_all(f)
except:
print 'Error! Wrong data format.'
return
for message in data:
try:
headers = message['headers']
body = eval(message['body'])
timestamp = headers['_publisher_timestamp']
timestamp = timestamp[:timestamp.find('.')]
data = [timestamp,body['event'],body]
p.process_event(data)
count = count + 1
except:
pass
print '\nProcessed %d events.' % count
开发者ID:ganga-devs,项目名称:gangamon,代码行数:33,代码来源:load_yaml.py
示例7: info
def info(self, path):
"""
Get info about a particular path.
"""
p = self.safePath(path)
info = {
'path': os.path.relpath(p, self.root),
'dir': os.path.isdir(p),
'file': os.path.isfile(p),
}
info['parent'] = os.path.join(info['path'], '..')
info['name'] = os.path.basename(p)
if info['dir']:
# directory
meta_filename = os.path.join(p, '_meta.yml')
if os.path.exists(meta_filename):
metadata = yaml.safe_load(open(meta_filename, 'rb'))
if metadata:
info.update(metadata)
else:
# file
if p.endswith('.md'):
metadata = yaml.safe_load_all(open(p, 'rb')).next()
if metadata and isinstance(metadata, dict):
info.update(metadata)
return info
开发者ID:iffy,项目名称:studytext,代码行数:26,代码来源:db.py
示例8: add_eyes
def add_eyes(self, views):
try:
with open(views, 'r') as f:
for view in yaml.safe_load_all(f):
self.eyes.extend(view)
except IOError as inst:
sys.stderr.write("Perspective file not found\n")
开发者ID:logan-j,项目名称:scraper,代码行数:7,代码来源:scrapetools.py
示例9: close
def close(self):
def get_report_path(report_id):
return os.path.join(self.report_dir, report_id)
report_filename = get_report_path(self.report_id)
try:
with open(report_filename) as fd:
g = yaml.safe_load_all(fd)
report_details = g.next()
except IOError:
raise e.ReportNotFound
dst_filename = report_file_name(report_details)
dst_path = os.path.join(self.archive_dir,
report_details['probe_cc'])
if not os.path.isdir(dst_path):
os.mkdir(dst_path)
dst_path = os.path.join(dst_path, dst_filename)
os.rename(report_filename, dst_path)
if not self.delayed_call.called:
self.delayed_call.cancel()
del self.reports[self.report_id]
开发者ID:isislovecruft,项目名称:ooni-backend,代码行数:25,代码来源:handlers.py
示例10: find_metadata
def find_metadata(self, gallery, lang):
"""Search for a gallery metadata file.
If there is an metadata file for the gallery, use that to determine
captions and the order in which images shall be displayed in the
gallery. You only need to list the images if a specific ordering or
caption is required. The metadata file is YAML-formatted, with field
names of
#
name:
caption:
order:
#
If a numeric order value is specified, we use that directly, otherwise
we depend on how PyYAML returns the information - which may or may not
be in the same order as in the file itself. Non-numeric ordering is not
supported. If no caption is specified, then we return an empty string.
Returns a string (l18n'd filename), list (ordering), dict (captions),
dict (image metadata).
"""
base_meta_path = os.path.join(gallery, "metadata.yml")
localized_meta_path = utils.get_translation_candidate(self.site.config,
base_meta_path, lang)
order = []
captions = {}
custom_metadata = {}
used_path = ""
if os.path.isfile(localized_meta_path):
used_path = localized_meta_path
elif os.path.isfile(base_meta_path):
used_path = base_meta_path
else:
return "", [], {}, {}
self.logger.debug("Using {0} for gallery {1}".format(
used_path, gallery))
with open(used_path, "r", encoding='utf-8-sig') as meta_file:
if yaml is None:
utils.req_missing(['PyYAML'], 'use metadata.yml files for galleries')
meta = yaml.safe_load_all(meta_file)
for img in meta:
# load_all and safe_load_all both return None as their
# final element, so skip it
if not img:
continue
if 'name' in img:
img_name = img.pop('name')
if 'caption' in img and img['caption']:
captions[img_name] = img.pop('caption')
if 'order' in img and img['order'] is not None:
order.insert(img.pop('order'), img_name)
else:
order.append(img_name)
custom_metadata[img_name] = img
else:
self.logger.error("no 'name:' for ({0}) in {1}".format(
img, used_path))
return used_path, order, captions, custom_metadata
开发者ID:scp93ch,项目名称:nikola,代码行数:60,代码来源:galleries.py
示例11: seminar_update_template
def seminar_update_template():
env = jinja2.Environment(
loader=jinja2.FileSystemLoader(searchpath='_seminars'),
trim_blocks=True,
lstrip_blocks=True,
undefined=jinja2.StrictUndefined
)
template = env.get_template('_template.md')
seminar_paths = [
seminar_file_entry.path
for seminar_file_entry
in os.scandir('_seminars')
if (
seminar_file_entry.is_file()
and os.path.splitext(seminar_file_entry.name)[1] == '.md'
and seminar_file_entry.name != '_template.md'
)
]
for seminar_path_current in seminar_paths:
# Load the existing seminar file
with open(seminar_path_current, encoding='utf-8') as f:
# Parse the YAML of the seminar
seminar = list(yaml.safe_load_all(f))[0]
# If we ever have more than one version, we'll need to check things here
assert seminar['version'] == 1
# Write it back using the template
seminar_rendered = template.render(seminar)
with open(seminar_path_current, encoding='utf-8', mode='w') as f:
f.write(seminar_rendered)
开发者ID:jayfo,项目名称:web-dub,代码行数:33,代码来源:seminar.py
示例12: ReadFileObject
def ReadFileObject(self, file_object):
"""Reads artifact definitions from a file-like object.
Args:
file_object: the file-like object to read from.
Yields:
Artifact definitions (instances of ArtifactDefinition).
Raises:
FormatError: if the format of the YAML artifact definition is not set
or incorrect.
"""
# TODO: add try, except?
yaml_generator = yaml.safe_load_all(file_object)
last_artifact_definition = None
for yaml_definition in yaml_generator:
try:
artifact_definition = self.ReadArtifactDefinitionValues(yaml_definition)
except errors.FormatError as exception:
error_location = u'At start'
if last_artifact_definition:
error_location = u'After: {0}'.format(last_artifact_definition.name)
raise errors.FormatError(u'{0} {1}'.format(error_location, exception))
yield artifact_definition
last_artifact_definition = artifact_definition
开发者ID:Onager,项目名称:artifacts,代码行数:29,代码来源:reader.py
示例13: ls
def ls(archive_dir, verbose):
for j in get_jobs(archive_dir):
job_dir = os.path.join(archive_dir, j)
summary = {}
try:
with file(os.path.join(job_dir, "summary.yaml")) as f:
g = yaml.safe_load_all(f)
for new in g:
summary.update(new)
except IOError as e:
if e.errno == errno.ENOENT:
print_debug_info(j, job_dir, archive_dir)
continue
else:
raise
print "{job} {status} {owner} {desc} {duration}s".format(
job=j,
owner=summary.get("owner", "-"),
desc=summary.get("description", "-"),
status=get_status(summary),
duration=int(summary.get("duration", 0)),
)
if verbose and "failure_reason" in summary:
print " {reason}".format(reason=summary["failure_reason"])
开发者ID:charpty,项目名称:teuthology,代码行数:25,代码来源:ls.py
示例14: load_from_stream
def load_from_stream(stream):
"""
Load configuration from a stream.
A stream could be a string or file descriptor
"""
return _merge_flow(yaml.safe_load_all(stream))
开发者ID:cea-hpc,项目名称:milkcheck,代码行数:7,代码来源:config.py
示例15: updatekeys
def updatekeys(ctx):
loglevel = logging.INFO
if ctx.verbose:
loglevel = logging.DEBUG
logging.basicConfig(
level=loglevel,
)
misc.read_config(ctx)
machines = [canonicalize_hostname(m) for m in ctx.machines]
if ctx.targets:
try:
with file(ctx.targets) as f:
g = yaml.safe_load_all(f)
for new in g:
if 'targets' in new:
for t in new['targets'].iterkeys():
machines.append(t)
except IOError as e:
raise argparse.ArgumentTypeError(str(e))
return scan_for_locks(ctx, machines)
开发者ID:alsall,项目名称:teuthology,代码行数:25,代码来源:lock.py
示例16: __init__
def __init__(self, filename, conf):
self.conf = conf
self.table = None
self.contents = None
self.dfn = None
self.final = False
self.spec = []
dfn_dir = os.path.abspath(os.path.dirname(filename))
with open(filename, 'r') as f:
definition = yaml.safe_load(f)
filter_specs = []
for dfn in definition['files']:
if isinstance(dfn, dict):
if 'file' in dfn:
filter_specs.append( (dfn['file'], dfn['level'], True) )
elif 'text' in dfn:
if 'title' in dfn:
filter_specs.append( ( (dfn['title'], dfn['text']), dfn['level'], False ) )
else:
filter_specs.append( ( dfn['text'], dfn['level'], False ) )
else:
raise Exception('[ERROR] [toc]: problem with {0} in {0}'.format(dfn, filename))
else:
filter_specs.append( (dfn, 1, True) )
all_objs = {}
self._first_source = definition['sources'][0]
for source in definition['sources']:
with open(os.path.join(dfn_dir, source), 'r') as f:
objs = yaml.safe_load_all(f)
for obj in objs:
all_objs[obj['file']] = obj
for fn, level, is_file in filter_specs:
if is_file is True:
try:
obj = all_objs[fn]
obj['level'] = level
self.spec.append(obj)
except KeyError:
raise Exception('[ERROR] [toc]: KeyError "{0}" in file: {1}'.format(fn, filename))
else:
# translation
if isinstance(fn, tuple):
self.spec.append( { 'name': fn[0],
'level': level,
'text': fn[1] } )
else:
self.spec.append( { 'name': None,
'level': level,
'text': fn } )
开发者ID:TylerBrock,项目名称:docs-tools,代码行数:60,代码来源:toc.py
示例17: test_yaml_load_sanity
def test_yaml_load_sanity(self):
data = list(yaml.safe_load_all("""
---
title: Test One
tags:
- inf1000
- oblig1
- arrays
text: |-
Hello
Cruel
World
---
title: Test Two
text: Testtext
"""))
self.assertEquals(len(data), 2)
self.assertEquals(data[0], {
'title': 'Test One',
'tags': ['inf1000', 'oblig1', 'arrays'],
'text': 'Hello\n\nCruel\n\nWorld'
})
self.assertEquals(data[1], {
'title': 'Test Two',
'text': 'Testtext'
})
开发者ID:devilry,项目名称:trix2,代码行数:28,代码来源:test_multiassignment_serialize.py
示例18: __init__
def __init__(self, report_file):
self._f = open(report_file)
self._skipped_line = 0
self.report_file = report_file
self.report_document = yaml.safe_load_all(self._f)
self.parse_header()
self.detect_nettest_format()
开发者ID:TheTorProject,项目名称:ooni-reader,代码行数:7,代码来源:report.py
示例19: check_file
def check_file(printer, path, config):
"""Check YAML file 'path' for differences.
:param printer: Where we report differences to.
:param str path: The YAML file to test.
:param dict config: Contains Kubernetes parsing and access configuration.
:return: Number of differences found.
"""
with open(path, 'r') as stream:
expected = yaml.safe_load_all(stream)
differences = 0
for data in expected:
# data can be None, e.g. in cases where the doc ends with a '---'
if not data:
continue
try:
for kube_obj in KubeObject.from_dict(data, config["namespace"]):
printer.add(path, kube_obj)
try:
running = kube_obj.get_from_cluster(config["kubeconfig"])
except subprocess.CalledProcessError as e:
printer.diff(path, Difference(e.output, None))
differences += 1
continue
for difference in diff("", kube_obj.data, running):
differences += 1
printer.diff(path, difference)
except Exception:
print("Failed parsing %s." % (path))
raise
return differences
开发者ID:weaveworks,项目名称:kubediff,代码行数:35,代码来源:_diff.py
示例20: __init__
def __init__(self, *args, **kwargs):
super(ArtifactsCollector, self).__init__(*args, **kwargs)
self.artifact_profile = self.session.LoadProfile("artifacts")
extra_definitions = [
open(x).read() for x in self.plugin_args.artifact_files]
extra_definitions.extend(self.plugin_args.definitions or [])
# Make a copy of the artifact registry.
if extra_definitions:
self.artifact_profile = self.artifact_profile.copy()
for definition in extra_definitions:
for definition_data in yaml.safe_load_all(definition):
self.artifact_profile.AddDefinition(definition_data)
self.seen = set()
self.supported_os = self.get_supported_os(self.session)
if self.supported_os is None:
raise plugin.PluginError(
"Unable to determine running environment.")
# Make sure the args make sense.
if self.plugin_args.output_path is None:
if self.plugin_args.copy_files:
raise plugin.PluginError(
"Can only copy files when an output file is specified.")
if self.plugin_args.create_timeline:
raise plugin.PluginError(
"Can only create timelines when an output file "
"is specified.")
开发者ID:google,项目名称:rekall,代码行数:31,代码来源:forensic_artifacts.py
注:本文中的yaml.safe_load_all函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论