本文整理汇总了Python中yaml.safe_dump_all函数的典型用法代码示例。如果您正苦于以下问题:Python safe_dump_all函数的具体用法?Python safe_dump_all怎么用?Python safe_dump_all使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了safe_dump_all函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: sanitise_report
def sanitise_report(report_file, semaphore):
match = re.search("^" + re.escape(settings.reports_directory) + "(.*)", report_file)
# read report file
report = Report(report_file)
report_header = report.header
report_header["report_file"] = match.group(1)
report_filename = os.path.split(report_file)[-1]
report_filename_sanitised = os.path.join(settings.sanitised_directory, report_filename)
if os.path.isfile(report_filename_sanitised):
log.info("Sanitised report name already exists, overwriting: %s" % report_filename_sanitised)
else:
log.info("New report file: %s" % report_filename_sanitised)
report_file_sanitised = open(report_filename_sanitised, "w")
safe_dump(report_header, report_file_sanitised, explicit_start=True, explicit_end=True)
safe_dump_all(report, report_file_sanitised, explicit_start=True, explicit_end=True, default_flow_style=False)
log.info("Moving original unsanitised file %s to archive" % report_file)
archive_report(report_file)
report_file_sanitised.close()
report.close()
os.remove(report_file)
semaphore.release()
开发者ID:andresazp,项目名称:simple_ooreport_viewer,代码行数:32,代码来源:sanitise.py
示例2: readpage
def readpage(page):
news=get_news(page)
if len(news)>0:
with open('meneame.yml', 'a') as outfile:
yaml.safe_dump_all(news, outfile, default_flow_style=False,allow_unicode=True, default_style=None)
outfile.write("---\n")
else:
print "SKIP"
开发者ID:santos82,项目名称:meneame.dump,代码行数:8,代码来源:meneame.py
示例3: read
def read(id):
url=STR+id
response = requests.get(url)
soup = bs4.BeautifulSoup(response.text,"lxml")
body=soup.select("div.news-body")
if not body or len(body)==0:
print "ERROR: "+url
return True
a=body[0]
cerrado=a.select("#a-va-"+id+" span.closed")
if len(cerrado)==0:
print "ABIERTA: "+url
return False
div=a.select("div.news-shakeit")
if len(div)==0:
print "SIN SHAKE: "+url
return False
status=div[0].attrs['class'][-1].split('-')[-1]
h1=soup.find("h1")
if not h1:
print "SIN h1: "+url
return False
dates=[int(d.attrs['data-ts'].strip()) for d in a.select("div.news-submitted span.ts")]
new={
'id': int(id),
'title': h1.get_text().strip(),
'url': h1.a.attrs["href"],
'author': a.select("div.news-submitted a")[0].attrs["href"].split("/")[-1],
'body': " ".join([t.string.strip() for t in a.find_all(text=True,recursive=False)]).strip(),
'sub': a.select("div.news-details span.tool a")[0].get_text().strip(),
'story': get_meta(soup,"property","og:url"),
'karma': int(a.select("#a-karma-"+id)[0].get_text().strip()),
'sent': dates[0],
'status': status,
'votes': {
'users': int(a.select("#a-usu-"+id)[0].get_text().strip()),
'anonymous': int(a.select("#a-ano-"+id)[0].get_text().strip()),
'negative':int(a.select("#a-neg-"+id)[0].get_text().strip())
}
}
if status == "published":
new['comments']=dates[1]
counter=a.select("span.comments-counter span.counter")
if len(counter)>0:
new['comments']=int(counter[0].get_text().strip())
else:
new['comments']=0
tags=[t.get_text().strip() for t in a.select("span.news-tags a") if len(t.get_text().strip())>0]
if len(tags)>0:
new['tags']=tags
with open(OUT, 'a') as outfile:
yaml.safe_dump_all([new], outfile, default_flow_style=False,allow_unicode=True, default_style=None)
outfile.write("---\n")
return True
开发者ID:santos82,项目名称:meneame.dump,代码行数:56,代码来源:meneame.all.py
示例4: main
def main(archive_file):
if not os.path.isfile(archive_file):
print "Archive file does not exist: " + archive_file
return 1
# we only support tar.gz and .gz
if archive_file.endswith("tar.gz"):
archive_tar = tarfile.open(archive_file)
for element in archive_tar:
f = archive_tar.extractfile(element)
fp, report_file = tempfile.mkstemp()
while True:
data = f.read()
if not data:
break
os.write(fp, data)
f.close()
elif archive_file.endswith(".gz"):
fp, report_file = tempfile.mkstemp()
f = gzip.open(archive_file, 'rb')
while True:
data = f.read()
if not data:
break
os.write(fp, data)
f.close()
report = Report(report_file, action="sanitise")
report_filename = generate_filename(report.header)
report_filename_sanitised = os.path.join(settings.sanitised_directory,
report_filename)
report.header['report_file'] = "%s/%s" % (report.header['probe_cc'],
report_filename)
report_file_sanitised = open(report_filename_sanitised, "w")
safe_dump(
report.header,
report_file_sanitised,
explicit_start=True,
explicit_end=True)
safe_dump_all(
report,
report_file_sanitised,
explicit_start=True,
explicit_end=True,
default_flow_style=False)
delete_existing_report_entries(report.header)
public_report_file = os.path.join(
settings.public_directory, report.header['probe_cc'], report_filename)
if os.path.isfile(public_report_file):
os.remove(public_report_file)
report_file_sanitised.close()
os.remove(report_file)
开发者ID:andresazp,项目名称:simple_ooreport_viewer,代码行数:56,代码来源:restore.py
示例5: main
def main():
parser = optparse.OptionParser()
parser.add_option("-H", "--hash", action="store",
help="verify that stdin hash given sha256 hash")
options, args = parser.parse_args()
if options.hash:
gen = process_package_with_hash(sys.stdin, options.hash)
else:
gen = process_package(sys.stdin)
yaml.safe_dump_all(gen, sys.stdout)
开发者ID:paulproteus,项目名称:debian-dedup,代码行数:10,代码来源:importpkg.py
示例6: write
def write(fname, *items):
try:
written = False
with closing(_open_userfile(fname, 'w')) as f:
if written:
f.write(SEPARATOR)
yaml.safe_dump_all(items, f)
except Exception as e:
print("Can't write filename", fname, e.message)
if PROPAGATE_EXCEPTIONS:
raise
开发者ID:dasbavaria,项目名称:echomesh,代码行数:11,代码来源:Yaml.py
示例7: make_adf
def make_adf(archive=None, canonical=False, out=None, pretty=False):
"""
>>> meta = Meta('zip', 'aes-256-ctr', created='now')
>>> archive = Archive('title', meta)
>>> cert = Certificate(chr(255)*16)
>>> make_adf([archive, cert], out=sys.stdout, pretty=True)
!archive
meta: !meta
cipher: aes-256-ctr
created: now
format: zip
title: title
--- !certificate
key: !!binary |
/////////////////////w==
>>> make_adf([archive, cert], out=sys.stdout)
!archive
meta: !meta {cipher: aes-256-ctr, created: now, format: zip}
title: title
--- !certificate
key: !!binary |
/////////////////////w==
"""
if not hasattr(archive, '__getitem__'):
archive = [archive]
if pretty:
out.write("--- ".join(map(pyaml.dump, archive)))
return
return yaml.safe_dump_all(archive, out, canonical=canonical)
开发者ID:V-Lamp,项目名称:longaccess-client,代码行数:30,代码来源:adf.py
示例8: CompileReport
def CompileReport(self, analysis_mediator):
"""Compiles a report of the analysis.
Args:
analysis_mediator: The analysis mediator object (instance of
AnalysisMediator).
Returns:
The analysis report (instance of AnalysisReport).
"""
report = event.AnalysisReport(self.NAME)
if self._output_format == u"yaml":
lines_of_text = []
lines_of_text.append(yaml.safe_dump_all(self._service_collection.services))
else:
lines_of_text = [u"Listing Windows Services"]
for service in self._service_collection.services:
lines_of_text.append(self._FormatServiceText(service))
# Separate services with a blank line.
lines_of_text.append(u"")
report.SetText(lines_of_text)
return report
开发者ID:hillu,项目名称:plaso,代码行数:25,代码来源:windows_services.py
示例9: test_deserializer_validate_existing_assignments
def test_deserializer_validate_existing_assignments(self):
assignment1 = coremodels.Assignment.objects.create(
title='Existing1', text='text1')
assignment2 = coremodels.Assignment.objects.create(
title='Existing2', text='text2')
duck1000tag = coremodels.Tag.objects.create(tag='duck1000')
assignment1.tags.add(duck1000tag)
assignment2.tags.add(duck1000tag)
deserializer = multiassignment_serialize.Deserializer(yaml.safe_dump_all([
{'id': assignment1.id, 'title': 'Updated1', 'text': 'updatedText1',
'tags': ['oblig1']},
{'title': 'New'},
{'id': assignment2.id, 'title': 'Updated2', 'text': 'updatedText2',
'tags': ['duck1000', 'oblig2', 'week3']},
]), course_tag='duck1000')
assignments_by_tag = {}
existing_assignments = deserializer._validate_existing_assignments(assignments_by_tag)
self.assertEquals(len(existing_assignments), 2)
self.assertEquals(
set(assignments_by_tag.keys()),
set(['duck1000', 'oblig1', 'oblig2', 'week3']))
self.assertEquals(
assignments_by_tag['duck1000'], [assignment2, assignment1])
self.assertEquals(
assignments_by_tag['oblig2'], [assignment2])
开发者ID:devilry,项目名称:trix2,代码行数:26,代码来源:test_multiassignment_serialize.py
示例10: pass_insert_host_yaml
def pass_insert_host_yaml(host, yaml_docs):
value = yaml.safe_dump_all(
yaml_docs,
# use multiline "block" style instead of {a:b} ("flow" style)
default_flow_style=False
)
pass_insert( host_to_pass_key(host), value )
开发者ID:pclewis,项目名称:dotfiles,代码行数:8,代码来源:pass.py
示例11: dump
def dump(arg):
"""Dump a sequence of dictionaries as YAML for editing.
"""
return yaml.safe_dump_all(
arg,
allow_unicode=True,
default_flow_style=False,
)
开发者ID:SerhatG,项目名称:nzbToMedia,代码行数:8,代码来源:edit.py
示例12: write_yaml
def write_yaml(input, filename):
with open(filename, "w") as f:
if isinstance(input, list):
f.write(yaml.safe_dump_all(input, default_flow_style=False))
elif isinstance(input, dict):
f.write(yaml.safe_dump(input, default_flow_style=False))
else:
raise Exception("cannot dump $s objects to yaml." % str(type(input)))
开发者ID:Bernadskaya,项目名称:docs,代码行数:8,代码来源:utils.py
示例13: test_deserializer_validate_existing_assignments_invalid
def test_deserializer_validate_existing_assignments_invalid(self):
assignment1 = coremodels.Assignment.objects.create(
title='Existing1', text='text1')
assignment1.tags.create(tag='duck1000')
deserializer = multiassignment_serialize.Deserializer(yaml.safe_dump_all([
{'id': assignment1.id},
]), course_tag='duck1000')
with self.assertRaises(multiassignment_serialize.DeserializerValidationErrors):
deserializer._validate_existing_assignments(assignments_by_tag={})
开发者ID:devilry,项目名称:trix2,代码行数:9,代码来源:test_multiassignment_serialize.py
示例14: main
def main(argv=sys.argv):
parser = arg_parser()
args = parser.parse_args()
print("Reading %s" % args.infile)
infile = open( args.infile )
d, report = csv2dict(infile)
print("Writing to %s" % args.outfile)
outfile = open( args.outfile, 'w' )
if args.fmt == 'json':
import json
outfile.write( json.dumps( d ) )
elif args.fmt == 'yaml':
import yaml
yaml.safe_dump_all( d, outfile, default_flow_style=False )
开发者ID:trmznt,项目名称:msaf,代码行数:20,代码来源:csv2dict.py
示例15: FormatArtifacts
def FormatArtifacts(self, artifacts):
"""Formats artifacts to desired output format.
Args:
artifacts (list[ArtifactDefinition]): artifact definitions.
Returns:
str: formatted string of artifact definition.
"""
# TODO: improve output formatting of yaml
artifact_definitions = [artifact.AsDict() for artifact in artifacts]
yaml_data = yaml.safe_dump_all(artifact_definitions)
return yaml_data
开发者ID:ForensicArtifacts,项目名称:artifacts,代码行数:13,代码来源:writer.py
示例16: test_deserializer_grouped_correctly
def test_deserializer_grouped_correctly(self):
deserializer = multiassignment_serialize.Deserializer(yaml.safe_dump_all([
{'id': 1, 'title': 'Existing1'},
{'title': 'New'},
{'id': 2, 'title': 'Existing2'},
]), course_tag='duck1000')
self.assertEquals(
deserializer.deserialized_assignments_with_id,
{1: {'id': 1, 'title': 'Existing1'}, 2: {'id': 2, 'title': 'Existing2'}}
)
self.assertEquals(
deserializer.deserialized_assignments_without_id,
[{'title': 'New'}]
)
开发者ID:devilry,项目名称:trix2,代码行数:14,代码来源:test_multiassignment_serialize.py
示例17: update_environments_yaml
def update_environments_yaml(self, key, val, provider='local'):
""" updates environments.yaml base file """
_env_yaml = os.path.expanduser("~/.juju/environments.yaml")
if os.path.exists(_env_yaml):
with open(_env_yaml) as f:
_env_yaml_raw = f.read()
env_yaml = yaml.load(_env_yaml_raw)
else:
raise ConfigException("~/.juju/environments.yaml unavailable, "
"is juju bootstrapped?")
if key in env_yaml['environments'][provider]:
env_yaml['environments'][provider][key] = val
with open(_env_yaml, 'w') as f:
_env_yaml_raw = yaml.safe_dump_all(env_yaml)
f.write(_env_yaml_raw)
开发者ID:gitter-badger,项目名称:cloud-installer,代码行数:15,代码来源:config.py
示例18: DumpMany
def DumpMany(objs):
"""Stringifies a sequence of Python objects to a multi-document YAML.
Args:
objs: An iterable of Python objects to convert to YAML.
Returns:
A multi-document YAML representation of the given objects.
"""
precondition.AssertIterableType(objs, object)
text = yaml.safe_dump_all(objs, default_flow_style=False, allow_unicode=True)
if compatibility.PY2:
text = text.decode("utf-8")
return text
开发者ID:google,项目名称:grr,代码行数:17,代码来源:yaml.py
示例19: main
def main(argv):
o = RewriteOptions()
try:
o.parseOptions(argv[1:])
except UsageError as e:
raise SystemExit(e)
docs = freeze(list(safe_load_all(stdin)))
if o["git-tag"] is not None:
tag = check_output(["git", "rev-parse", "--short", o["git-tag"]]).strip()
docs = rewrite_tags(docs, tag)
elif o["tag"] is not None:
docs = rewrite_tags(docs, o["tag"])
if o["no-volumes"]:
docs = stub_all_volumes(docs)
stdout.write(safe_dump_all(thaw(docs)))
开发者ID:LeastAuthority,项目名称:leastauthority.com,代码行数:19,代码来源:rewrite-objects.py
示例20: test_deserializer_sync
def test_deserializer_sync(self):
assignment1 = coremodels.Assignment.objects.create(
title='Existing1', text='text1')
assignment2 = coremodels.Assignment.objects.create(
title='Existing2', text='text2')
duck1000tag = coremodels.Tag.objects.create(tag='duck1000')
assignment1.tags.add(duck1000tag)
assignment2.tags.add(duck1000tag)
deserializer = multiassignment_serialize.Deserializer(yaml.safe_dump_all([
{'id': assignment1.id, 'title': 'Updated1', 'text': 'updatedText1',
'tags': ['oblig1']},
{'title': 'New1', 'text': 'newText1'},
{'title': 'New2', 'text': 'newText2',
'tags': ['duck1000', 'oblig1'],
'solution': 'newSolution'},
{'id': assignment2.id, 'title': 'Updated2', 'text': 'updatedText2',
'tags': ['duck1000', 'oblig2'],
'solution': 'updatedSolution'},
]), course_tag='duck1000')
deserializer.sync()
assignment1 = coremodels.Assignment.objects.get(id=assignment1.id)
assignment2 = coremodels.Assignment.objects.get(id=assignment2.id)
newassignment1 = coremodels.Assignment.objects.get(title='New1')
newassignment2 = coremodels.Assignment.objects.get(title='New2')
self.assertEquals(assignment1.title, 'Updated1')
self.assertEquals(assignment1.solution, '')
self.assertEquals(assignment2.title, 'Updated2')
self.assertEquals(assignment2.solution, 'updatedSolution')
self.assertEquals(
set([tagobject.tag for tagobject in assignment1.tags.all()]),
set(['duck1000', 'oblig1']))
self.assertEquals(
set([tagobject.tag for tagobject in assignment2.tags.all()]),
set(['duck1000', 'oblig2']))
self.assertEquals(
set([tagobject.tag for tagobject in newassignment1.tags.all()]),
set(['duck1000']))
self.assertEquals(
set([tagobject.tag for tagobject in newassignment2.tags.all()]),
set(['duck1000', 'oblig1']))
开发者ID:devilry,项目名称:trix2,代码行数:42,代码来源:test_multiassignment_serialize.py
注:本文中的yaml.safe_dump_all函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论