本文整理汇总了Python中yaml.load函数的典型用法代码示例。如果您正苦于以下问题:Python load函数的具体用法?Python load怎么用?Python load使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了load函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: process_message
def process_message(self, peer, mailfrom, rcpttos, data):
if savemsg:
filename = "savedmsg/%s-%d.eml" % (datetime.now().strftime("%Y%m%d%H%M%S"), self.no)
f = open(filename, "w")
f.write(data)
f.close
if "X-Arf: yes" in data:
mail = email.message_from_string(data)
for part in mail.walk():
ctypes = part.get_params(None, "Content-Type")
if ctypes and len(ctypes) > 2:
if ctypes[0] == ("text/plain", "") and ctypes[2] == ("name", "report.txt"):
payload = part.get_payload()
xarf = load(payload, Loader=Loader)
handle_xarf(xarf)
dtypes = part.get_params(None, "Content-Disposition")
if dtypes and dtypes[1] == ("filename", "report.txt"):
payload = b64decode(part.get_payload())
xarf = load(payload, Loader=Loader)
handle_xarf(xarf)
print "Email received at %s" % datetime.now().strftime("%H:%M:%S")
self.no += 1
开发者ID:funzoneq,项目名称:x-arf,代码行数:26,代码来源:smtp.py
示例2: test_yaml_representation_has_all_expected_fields
def test_yaml_representation_has_all_expected_fields(self):
"""Verify that the YAML representation of reference props is ok."""
prop = properties.ReferenceProperty(
'name', references.Reference(
'5f2c9a1d-1113-49f1-9d1d-29aaa4a520b0', None, None))
string = yaml.dump(prop)
data = yaml.load(string)
self.assertTrue(isinstance(data, dict))
self.assertEqual(data['uuid'], '5f2c9a1d-1113-49f1-9d1d-29aaa4a520b0')
self.assertTrue(not 'service' in data)
self.assertTrue(not 'ref' in data)
prop = properties.ReferenceProperty(
'name', references.Reference(
'5f2c9a1d-1113-49f1-9d1d-29aaa4a520b0', 'issues', None))
string = yaml.dump(prop)
data = yaml.load(string)
self.assertTrue(isinstance(data, dict))
self.assertEqual(data['uuid'], '5f2c9a1d-1113-49f1-9d1d-29aaa4a520b0')
self.assertEqual(data['service'], 'issues')
self.assertTrue(not 'ref' in data)
prop = properties.ReferenceProperty(
'name', references.Reference(
'5f2c9a1d-1113-49f1-9d1d-29aaa4a520b0',
'issues', 'master'))
string = yaml.dump(prop)
data = yaml.load(string)
self.assertTrue(isinstance(data, dict))
self.assertEqual(data['uuid'], '5f2c9a1d-1113-49f1-9d1d-29aaa4a520b0')
self.assertEqual(data['service'], 'issues')
self.assertEqual(data['ref'], 'master')
开发者ID:Jannis,项目名称:python-consonant,代码行数:33,代码来源:properties_tests.py
示例3: get_config
def get_config(queue_names,
yaml_filename='/etc/tada/tada.conf',
hiera_filename='/etc/tada/hiera.yaml',
validate=False):
"""Read multi-queue config from yaml_filename. Validate its
contents. Insure queue_names are all in the list of named queues."""
try:
cfg = yaml.load(open(yaml_filename))
except:
raise Exception('ERROR: Could not read data-queue config file "{}"'
.format(yaml_filename))
try:
cfg.update(yaml.load(open(hiera_filename)))
except Exception as err:
raise Exception('ERROR: Could not read data-queue config file "{}"; {}'
.format(hiera_filename, str(err)))
if validate:
validate_config(cfg, qnames=queue_names, fname=yaml_filename)
#!lut = get_config_lut(cfg)
#return dict([[q['name'], q] for q in config['queues']])
lut = cfg
if validate:
missing = set(queue_names) - set(lut.keys())
else:
missing = set()
if len(missing) > 0:
raise Exception(
'ERROR: Config file "{}" does not contain named queues: {}'
.format(yaml_filename, missing))
#return lut, cfg['dirs']
logging.debug('get_config got: {}'.format(lut))
return lut, dict()
开发者ID:pothiers,项目名称:tada,代码行数:33,代码来源:config.py
示例4: test_cleaner
def test_cleaner(self):
c = util.Cleaner(hide_keys=['pass'], hide_values=['xyz', 'zyx'])
_in = yaml.load("""
pass: alma
password: xyz
public: yaay
stuff:
-
secret: zyx
-
- zyx
- alma
""")
_out = yaml.load("""
pass: XXX
password: XXX
public: yaay
stuff:
-
secret: XXX
-
- XXX
- alma
""")
obfuscated = c.deep_copy(_in)
self.assertEqual(obfuscated, _out)
开发者ID:novakadam94,项目名称:util,代码行数:26,代码来源:general_tests.py
示例5: setUpClass
def setUpClass(self):
valid_contour_config = """
type: contour
results_indices:
- !!python/tuple [0, 0]
lats:
range_min: -20
range_max: 20
range_step: 1
lons:
range_min: -20
range_max: 20
range_step: 1
output_name: wrf_bias_compared_to_knmi
"""
self.valid_contour = yaml.load(valid_contour_config)
missing_keys_contour_config = """
type: contour
"""
self.missing_keys_contour = yaml.load(missing_keys_contour_config)
self.required_contour_keys = set([
'results_indices',
'lats',
'lons',
'output_name'
])
开发者ID:CWSL,项目名称:climate,代码行数:28,代码来源:test_config_parsing.py
示例6: load_from_dir
def load_from_dir(self, directory):
with open("{}/yaml/venues.yaml".format(directory), "r") as f:
venue_dict = yaml.load(f, Loader=Loader)
for acronym, name_str in venue_dict.items():
name, venue_type = name_str.split(":")
self.venues[acronym] = {
"name": name,
"is_acl": (venue_type == "ACL"),
"is_toplevel": False,
"slug": slugify(acronym),
"type": venue_type,
"years": set(),
"volumes": [],
}
with open("{}/yaml/venues_letters.yaml".format(directory), "r") as f:
self.letters = yaml.load(f, Loader=Loader)
for letter, acronym in self.letters.items():
self.venues[acronym]["is_toplevel"] = True
self.venues[acronym]["main_letter"] = letter
with open("{}/yaml/venues_joint_map.yaml".format(directory), "r") as f:
map_dict = yaml.load(f, Loader=Loader)
for id_, joint in map_dict.items():
if isinstance(joint, str):
joint = [joint]
self.joint_map[id_] = joint
开发者ID:WING-NUS,项目名称:acl,代码行数:25,代码来源:venues.py
示例7: main
def main():
# Connection Database
conf_file_path = sys.path[0]+"/conf/crawl.conf"
conf_file = open(conf_file_path)
conf = yaml.load(conf_file)
host = conf['host']
port = conf['port']
user = conf['user']
password = conf['password']
database = conf['conf_database']
tablename = conf['conf_table']
# Read initial_sql_table
sql_file_path = sys.path[0]+"/conf/sql.ini"
sql_file = open(sql_file_path)
sql = yaml.load(sql_file)
# SQL Section
comment_table = sql['top_api_shop_order']
try:
itemstage = sys.argv[1]
except:
itemstage = 10
projectlist = getTask(host,port,user,password,database,tablename,itemstage)
for project in projectlist:
do_run(host,port,user,password,project[0],project[1],project[2],project[3],comment_table)
开发者ID:ricklovelisa,项目名称:python_crawler,代码行数:28,代码来源:10_api_order.py
示例8: test_unsafe
def test_unsafe(self):
dummy = Dummy()
with self.assertRaises(yaml.representer.RepresenterError):
yaml.dump_all([dummy])
with self.assertRaises(yaml.representer.RepresenterError):
yaml.dump(dummy, Dumper=yDumper)
# reverse monkey patch and try again
monkey_patch_pyyaml_reverse()
with tempfile.TemporaryFile(suffix='.yaml') as f:
yaml.dump_all([dummy], stream=f)
f.seek(0) # rewind
doc_unsafe = yaml.load(f)
self.assertTrue(type(doc_unsafe) is Dummy)
monkey_patch_pyyaml()
with self.assertRaises(yaml.constructor.ConstructorError):
f.seek(0) # rewind
safe_yaml_load(f)
with self.assertRaises(yaml.constructor.ConstructorError):
f.seek(0) # rewind
yaml.load(f)
开发者ID:DataDog,项目名称:dd-agent,代码行数:27,代码来源:test_utils_yaml.py
示例9: test_checkbox_value
def test_checkbox_value(self):
attrs = '''
editable:
storage:
osd_pool_size:
description: desc
label: OSD Pool Size
type: checkbox
value: true
weight: 80
'''
self.assertNotRaises(errors.InvalidData,
AttributesValidator.validate_editable_attributes,
yaml.load(attrs))
attrs = '''
editable:
storage:
osd_pool_size:
description: desc
label: OSD Pool Size
type: checkbox
value: 'x'
weight: 80
'''
self.assertRaises(errors.InvalidData,
AttributesValidator.validate_editable_attributes,
yaml.load(attrs))
开发者ID:kansuke4649,项目名称:fuel-web,代码行数:29,代码来源:test_attributes_validator.py
示例10: evaluate
def evaluate(definition, args, account_info, force: bool):
# extract Senza* meta information
info = definition.pop("SenzaInfo")
info["StackVersion"] = args.version
template = yaml.dump(definition, default_flow_style=False)
definition = evaluate_template(template, info, [], args, account_info)
definition = yaml.load(definition)
components = definition.pop("SenzaComponents", [])
# merge base template with definition
BASE_TEMPLATE.update(definition)
definition = BASE_TEMPLATE
# evaluate all components
for component in components:
componentname, configuration = named_value(component)
configuration["Name"] = componentname
componenttype = configuration["Type"]
componentfn = get_component(componenttype)
if not componentfn:
raise click.UsageError('Component "{}" does not exist'.format(componenttype))
definition = componentfn(definition, configuration, args, info, force)
# throw executed template to templating engine and provide all information for substitutions
template = yaml.dump(definition, default_flow_style=False)
definition = evaluate_template(template, info, components, args, account_info)
definition = yaml.load(definition)
return definition
开发者ID:ehartung,项目名称:senza,代码行数:34,代码来源:cli.py
示例11: run_condor_jobs
def run_condor_jobs(c, config_file, subject_list_file, p_name):
'''
'''
# Import packages
import commands
from time import strftime
try:
sublist = yaml.load(open(os.path.realpath(subject_list_file), 'r'))
except:
raise Exception ("Subject list is not in proper YAML format. Please check your file")
cluster_files_dir = os.path.join(os.getcwd(), 'cluster_files')
subject_bash_file = os.path.join(cluster_files_dir, 'submit_%s.condor' % str(strftime("%Y_%m_%d_%H_%M_%S")))
f = open(subject_bash_file, 'w')
print >>f, "Executable = /usr/bin/python"
print >>f, "Universe = vanilla"
print >>f, "transfer_executable = False"
print >>f, "getenv = True"
print >>f, "log = %s" % os.path.join(cluster_files_dir, 'c-pac_%s.log' % str(strftime("%Y_%m_%d_%H_%M_%S")))
sublist = yaml.load(open(os.path.realpath(subject_list_file), 'r'))
for sidx in range(1,len(sublist)+1):
print >>f, "error = %s" % os.path.join(cluster_files_dir, 'c-pac_%s.%s.err' % (str(strftime("%Y_%m_%d_%H_%M_%S")), str(sidx)))
print >>f, "output = %s" % os.path.join(cluster_files_dir, 'c-pac_%s.%s.out' % (str(strftime("%Y_%m_%d_%H_%M_%S")), str(sidx)))
print >>f, "arguments = \"-c 'import CPAC; CPAC.pipeline.cpac_pipeline.run( ''%s'',''%s'',''%s'',''%s'',''%s'',''%s'',''%s'')\'\"" % (str(config_file), subject_list_file, str(sidx), c.maskSpecificationFile, c.roiSpecificationFile, c.templateSpecificationFile, p_name)
print >>f, "queue"
f.close()
#commands.getoutput('chmod +x %s' % subject_bash_file )
print commands.getoutput("condor_submit %s " % (subject_bash_file))
开发者ID:FCP-INDI,项目名称:C-PAC,代码行数:35,代码来源:cpac_runner.py
示例12: _install_container_bcbio_system
def _install_container_bcbio_system(datadir):
"""Install limited bcbio_system.yaml file for setting core and memory usage.
Adds any non-specific programs to the exposed bcbio_system.yaml file, only
when upgrade happening inside a docker container.
"""
base_file = os.path.join(datadir, "config", "bcbio_system.yaml")
if not os.path.exists(base_file):
return
expose_file = os.path.join(datadir, "galaxy", "bcbio_system.yaml")
expose = set(["memory", "cores", "jvm_opts"])
with open(base_file) as in_handle:
config = yaml.load(in_handle)
if os.path.exists(expose_file):
with open(expose_file) as in_handle:
expose_config = yaml.load(in_handle)
else:
expose_config = {"resources": {}}
for pname, vals in config["resources"].iteritems():
expose_vals = {}
for k, v in vals.iteritems():
if k in expose:
expose_vals[k] = v
if len(expose_vals) > 0 and pname not in expose_config["resources"]:
expose_config["resources"][pname] = expose_vals
with open(expose_file, "w") as out_handle:
yaml.safe_dump(expose_config, out_handle, default_flow_style=False, allow_unicode=False)
return expose_file
开发者ID:GetBen,项目名称:bcbio-nextgen,代码行数:28,代码来源:install.py
示例13: test_disks_flag
def test_disks_flag(self):
# specifying a EBS mount or PD mount is only valid for EC2/Euca/GCE, so
# fail on a cluster deployment.
argv = self.cluster_argv[:] + ["--disks", "ABCDFEG"]
self.assertRaises(BadConfigurationException, ParseArgs, argv, self.function)
# if we get a --disk flag, fail if it's not a dict (after base64, yaml load)
bad_disks_layout = yaml.load("""
public1,
""")
base64ed_bad_disks = base64.b64encode(yaml.dump(bad_disks_layout))
cloud_argv1 = self.cloud_argv[:] + ["--disks", base64ed_bad_disks]
self.assertRaises(BadConfigurationException, ParseArgs, cloud_argv1,
self.function)
# passing in a dict should be fine, and result in us seeing the same value
# for --disks that we passed in.
disks = {'public1' : 'vol-ABCDEFG'}
good_disks_layout = yaml.load("""
public1 : vol-ABCDEFG
""")
base64ed_good_disks = base64.b64encode(yaml.dump(good_disks_layout))
cloud_argv2 = self.cloud_argv[:] + ["--disks", base64ed_good_disks]
actual = ParseArgs(cloud_argv2, self.function).args
self.assertEquals(disks, actual.disks)
开发者ID:ArielSnir,项目名称:appscale-tools,代码行数:25,代码来源:test_parse_args.py
示例14: test_extra_options
def test_extra_options(self):
device = NewDevice(os.path.join(os.path.dirname(__file__), '../devices/kvm01.yaml'))
kvm_yaml = os.path.join(os.path.dirname(__file__), 'sample_jobs/kvm-inline.yaml')
with open(kvm_yaml) as sample_job_data:
job_data = yaml.load(sample_job_data)
device['actions']['boot']['methods']['qemu']['parameters']['extra'] = yaml.load("""
- -smp
- 1
- -global
- virtio-blk-device.scsi=off
- -device virtio-scsi-device,id=scsi
- --append "console=ttyAMA0 root=/dev/vda rw"
""")
self.assertIsInstance(device['actions']['boot']['methods']['qemu']['parameters']['extra'][1], int)
parser = JobParser()
job = parser.parse(yaml.dump(job_data), device, 4212, None, None, None,
output_dir='/tmp/')
job.validate()
boot_image = [action for action in job.pipeline.actions if action.name == 'boot_image_retry'][0]
boot_qemu = [action for action in boot_image.internal_pipeline.actions if action.name == 'boot_qemu_image'][0]
qemu = [action for action in boot_qemu.internal_pipeline.actions if action.name == 'execute-qemu'][0]
self.assertIsInstance(qemu.sub_command, list)
[self.assertIsInstance(item, str) for item in qemu.sub_command]
self.assertIn('virtio-blk-device.scsi=off', qemu.sub_command)
self.assertIn('1', qemu.sub_command)
self.assertNotIn(1, qemu.sub_command)
开发者ID:BayLibre,项目名称:lava-dispatcher,代码行数:26,代码来源:test_kvm.py
示例15: main
def main():
logging.basicConfig(format='%(asctime)s %(levelname)s: %(message)s', level=logging.INFO)
logging.getLogger('requests').setLevel(logging.WARNING)
setup_signal_handlers()
# Patroni reads the configuration from the command-line argument if it exists, and from the environment otherwise.
use_env = False
use_file = (len(sys.argv) >= 2 and os.path.isfile(sys.argv[1]))
if not use_file:
config_env = os.environ.get(Patroni.PATRONI_CONFIG_VARIABLE)
use_env = config_env is not None
if not use_env:
print('Usage: {0} config.yml'.format(sys.argv[0]))
print('\tPatroni may also read the configuration from the {} environment variable'.
format(Patroni.PATRONI_CONFIG_VARIABLE))
return
if use_file:
with open(sys.argv[1], 'r') as f:
config = yaml.load(f)
elif use_env:
config = yaml.load(config_env)
patroni = Patroni(config)
try:
patroni.run()
except KeyboardInterrupt:
pass
finally:
patroni.api.shutdown()
patroni.postgresql.stop(checkpoint=False)
patroni.dcs.delete_leader()
开发者ID:jankeirse,项目名称:patroni,代码行数:32,代码来源:__init__.py
示例16: get_configuration
def get_configuration(default_filename: str = 'defaults.yml', user_filename: str = 'config.yml') -> ConfigTree:
"""
gets the current configuration, as specified by YAML files
:param default_filename: name of the default settings file (relative to :file:`configparser.py`)
:param user_filename: name of the user settings file (relative to :file:`configparser.py`)
:return: settings tree
"""
# read defaults
with open(default_filename, 'r') as file:
defaults = yaml.load(file)
logger.info("Successfully parsed {} as default configuration".format(default_filename))
with open(user_filename, 'r') as file:
user_config = yaml.load(file)
logger.info("Successfully parsed {} as user configuration".format(user_filename))
# apply user config over Defaults
configuration = update_settings_tree(base=defaults, update=user_config)
# parse MQTT path templates
for path_name in configuration.MQTT.Path:
path_template = configuration.MQTT.Path[path_name]
path = path_template.format(prefix=configuration.MQTT.prefix, sys_name=configuration.sys_name)
configuration.MQTT.Path[path_name] = path
return configuration
开发者ID:Yottabits,项目名称:DotStar,代码行数:29,代码来源:configparser.py
示例17: __read_over
def __read_over(self, overstate):
'''
Read in the overstate file
'''
if overstate:
with salt.utils.fopen(overstate) as fp_:
try:
# TODO Use render system
return self.__sort_stages(yaml.load(fp_))
except Exception:
return {}
if self.env not in self.opts['file_roots']:
return {}
for root in self.opts['file_roots'][self.env]:
fn_ = os.path.join(
root,
self.opts.get('overstate', 'overstate.sls')
)
if not os.path.isfile(fn_):
continue
with salt.utils.fopen(fn_) as fp_:
try:
# TODO Use render system
return self.__sort_stages(yaml.load(fp_))
except Exception:
return {}
return {}
开发者ID:inthecloud247,项目名称:salt,代码行数:27,代码来源:overstate.py
示例18: get_jinja_vars
def get_jinja_vars(self):
# order for per-project variables (each overrides the previous):
# 1. /etc/kolla/globals.yml and passwords.yml
# 2. config/all.yml
# 3. config/<project>/defaults/main.yml
with open(file_utils.find_config_file('passwords.yml'), 'r') as gf:
global_vars = yaml.load(gf)
with open(file_utils.find_config_file('globals.yml'), 'r') as gf:
global_vars.update(yaml.load(gf))
all_yml_name = os.path.join(self.config_dir, 'all.yml')
jvars = yaml.load(jinja_utils.jinja_render(all_yml_name, global_vars))
jvars.update(global_vars)
for proj in self.get_projects():
proj_yml_name = os.path.join(self.config_dir, proj,
'defaults', 'main.yml')
if os.path.exists(proj_yml_name):
proj_vars = yaml.load(jinja_utils.jinja_render(proj_yml_name,
jvars))
jvars.update(proj_vars)
else:
LOG.warn('path missing %s' % proj_yml_name)
# override node_config_directory to empty
jvars.update({'node_config_directory': ''})
return jvars
开发者ID:alchemy-solutions,项目名称:kolla-mesos,代码行数:28,代码来源:deploy.py
示例19: _cmd_config_set
def _cmd_config_set(self, argument):
"""
set a configvalue.
syntax for argument: [network=net] [channel=chan] module.setting newvalue
"""
#how this works:
#args[x][:8] is checked for network= or channel=. channel= must come after network=
#args[x][8:] is the network/channelname without the prefix
#" ".join(args[x:]) joins all arguments after network= and channel= to a string from word x to the end of input
args=argument.split(" ")
if len(args)>=4 and len(args[0])>=8 and len(args[1])>=8 and args[0][:8]=="network=" and args[1][:8]=="channel=":
try:
(module, setting)=args[2].split(".", 1)
self.parent.getServiceNamed("config").set(setting, yaml.load(" ".join(args[3:])), module, args[0][8:], args[1][8:])
return self.parent.getServiceNamed("config").get(setting, "[unset]", module, args[0][8:], args[1][8:])
except ValueError:
return "Error: your setting is not in the module.setting form"
elif len(args)>=3 and len(args[0])>=8 and args[0][:8]=="network=":
try:
(module, setting)=args[1].split(".", 1)
self.parent.getServiceNamed("config").set(setting, yaml.load(" ".join(args[2:])), module, args[0][8:])
return self.parent.getServiceNamed("config").get(setting, "[unset]", module, args[0][8:])
except ValueError:
return "Error: your setting is not in the module.setting form"
elif len(argument):
try:
(module, setting)=args[0].split(".", 1)
self.parent.getServiceNamed("config").set(args[0], yaml.load(" ".join(args[1:])), module)
return self.parent.getServiceNamed("config").get(setting, "[unset]", module)
except ValueError:
return "Error: your setting is not in the module.setting form"
else:
return "config set [network=networkname] [channel=#somechannel] setting value"
开发者ID:otfbot,项目名称:otfbot,代码行数:33,代码来源:control.py
示例20: _main
def _main(args, config):
logging.setup_logging(config)
fp = config.ircbot_channel_config
if fp:
fp = os.path.expanduser(fp)
if not os.path.exists(fp):
raise ElasticRecheckException(
"Unable to read layout config file at %s" % fp)
else:
raise ElasticRecheckException(
"Channel Config must be specified in config file.")
channel_config = ChannelConfig(yaml.load(open(fp)))
msgs = MessageConfig(yaml.load(open(fp)))
if not args.noirc:
bot = RecheckWatchBot(
channel_config.channels,
config=config)
else:
bot = None
recheck = RecheckWatch(
bot,
channel_config,
msgs,
config=config,
commenting=not args.nocomment,
)
recheck.start()
if not args.noirc:
bot.start()
开发者ID:openstack-infra,项目名称:elastic-recheck,代码行数:34,代码来源:bot.py
注:本文中的yaml.load函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论