本文整理汇总了Python中yaml.safe_load函数的典型用法代码示例。如果您正苦于以下问题:Python safe_load函数的具体用法?Python safe_load怎么用?Python safe_load使用的例子?那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了safe_load函数的20个代码示例,这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞,您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: process_remote_sources
def process_remote_sources(raw_config, environment=None):
"""Stage remote package sources and merge in remote configs.
Args:
raw_config (str): the raw stacker configuration string.
environment (dict, optional): any environment values that should be
passed to the config
Returns:
str: the raw stacker configuration string
"""
config = yaml.safe_load(raw_config)
if config and config.get('package_sources'):
processor = SourceProcessor(
sources=config['package_sources'],
stacker_cache_dir=config.get('stacker_cache_dir')
)
processor.get_package_sources()
if processor.configs_to_merge:
for i in processor.configs_to_merge:
logger.debug("Merging in remote config \"%s\"", i)
remote_config = yaml.safe_load(open(i))
config = merge_map(remote_config, config)
# Call the render again as the package_sources may have merged in
# additional environment lookups
if not environment:
environment = {}
return render(str(config), environment)
return raw_config
开发者ID:krashproof,项目名称:stacker,代码行数:32,代码来源:__init__.py
示例2: configure
def configure(self):
if os.path.isfile(installation_path + "/conf/user_data.yaml.tmpl"):
userdatayaml = installation_path + "/user-data/" + self.username
if os.path.isfile(userdatayaml):
userdatayaml_data_stream = open(userdatayaml, "r")
yaml_parsed_userdata = yaml.safe_load(userdatayaml_data_stream)
userdatayaml_data_stream.close()
myversion = yaml_parsed_userdata.get("PHP")
backend_config_file = installation_path + "/conf/backends.yaml"
backend_data_yaml = open(backend_config_file, "r")
backend_data_yaml_parsed = yaml.safe_load(backend_data_yaml)
backend_data_yaml.close()
if "PHP" in backend_data_yaml_parsed:
php_backends_dict = backend_data_yaml_parsed["PHP"]
php_path = php_backends_dict.get(myversion)
php_profile_set(self.username, php_path)
path_to_socket = php_path + "/var/run/" + self.username + ".sock"
if os.path.islink("/opt/fpmsockets/" + self.username + ".sock"):
os.remove("/opt/fpmsockets/" + self.username + ".sock")
os.symlink(path_to_socket, "/opt/fpmsockets/" + self.username + ".sock")
else:
os.symlink(path_to_socket, "/opt/fpmsockets/" + self.username + ".sock")
else:
print("ERROR:: PHP Backends missing")
else:
subprocess.call("cp " + installation_path + "/conf/user_data.yaml.tmpl " + userdatayaml, shell=True)
cpuser_uid = pwd.getpwnam(self.username).pw_uid
cpuser_gid = grp.getgrnam(self.username).gr_gid
os.chown(userdatayaml, cpuser_uid, cpuser_gid)
os.chmod(userdatayaml, 0660)
self.configure()
else:
sys.exit(0)
开发者ID:xlightwaverx,项目名称:nDeploy,代码行数:33,代码来源:apache_php_config_generator.py
示例3: _validate_output
def _validate_output(serial_str):
try:
yaml.safe_load(StringIO(serial_str))
except Exception:
return False
else:
return True
开发者ID:gongfacun,项目名称:django,代码行数:7,代码来源:tests.py
示例4: test_configuration_with_binary_strings
def test_configuration_with_binary_strings():
"""
Regression test: serialization was failing on binary strings
"""
import yaml
obj = '\xaa\xbb\x00\xff\xff\x00ABC'
assert yaml.load(yaml.dump(obj)) == obj
assert yaml.safe_load(yaml.safe_dump(obj)) == obj
obj = {'blob': '\xaa\xbb\x00\xff\xff\x00ABC'}
assert yaml.load(yaml.dump(obj)) == obj
assert yaml.safe_load(yaml.safe_dump(obj)) == obj
obj = {
'function': 'jobcontrol.utils.testing:job_simple_echo',
'title': None,
'notes': None,
# 'args': ('\xaa\xbb\x00\xff\xff\x00ABC',),
'args': '\xaa\xbb\x00\xff\xff\x00ABC',
'dependencies': [],
'kwargs': {},
'id': 'f974e89f-4ae3-40cc-8316-b78e42bd5cc8',
}
dump(obj)
开发者ID:vortec,项目名称:jobcontrol,代码行数:26,代码来源:test_configuration.py
示例5: __init__
def __init__(self, filename=None, data=None, fd=None):
super(YamlParser, self).__init__()
if fd:
self.parsed = yaml.safe_load(fd)
self.fd = fd
try:
self.filename = fd.name
except AttributeError:
self.filename = None
elif filename:
try:
self.parsed = yaml.safe_load(open(filename, "rb")) or OrderedYamlDict()
except IOError as e:
if e.errno == errno.EACCES:
# Specifically catch access denied errors, this usually indicates the
# user wanted to read the file, and it existed, but they lacked the
# permissions.
raise IOError(e)
else:
self.parsed = OrderedYamlDict()
except OSError:
self.parsed = OrderedYamlDict()
self.filename = filename
elif data is not None:
fd = StringIO.StringIO(data)
self.parsed = yaml.safe_load(fd)
self.filename = filename
else:
raise Error("Filename not specified.")
开发者ID:andrewseidl,项目名称:grr,代码行数:34,代码来源:config_lib.py
示例6: test_deprecation
def test_deprecation(spec_fixture, workspace_manager_fixture, # noqa
test_workspace, tmpdir):
"""Verify execution runs with deprecated option """
my_temp_dir = tmpdir.mkdir("tmp")
deprecated_output = my_temp_dir.join("deprecated_output.yml")
deprecated_input_string = \
['example', '--deprecated-way', 'TestingValue', '--dry-run',
'-o', str(deprecated_output)]
output = my_temp_dir.join("output.yml")
input_string = \
['example', '--new-way', 'TestingValue', '--dry-run',
'-o', str(output)]
spec_manager = api.SpecManager()
spec_manager.register_spec(spec_fixture)
workspace_manager_fixture.activate(test_workspace.name)
spec_manager.run_specs(args=deprecated_input_string)
spec_manager.run_specs(args=input_string)
with open(deprecated_output.strpath) as fp:
deprecated_yml = yaml.safe_load(fp)["provision"]
with open(output.strpath) as fp:
new_yml = yaml.safe_load(fp)["provision"]
assert deprecated_yml.get('new', None).get('way', None) == 'TestingValue'
assert new_yml.get('new', None).get('way', None) == 'TestingValue'
开发者ID:aopincar,项目名称:InfraRed,代码行数:32,代码来源:test_execute.py
示例7: render_template
def render_template(path, week=None, **kwargs):
with open('out/report.yml') as r:
report = yaml.safe_load(r)
with open('bloggers.yml') as f:
users = yaml.safe_load(f)
if week:
week = parse(week, default=START)
else:
week = START
week = (week - START).days / 7
week_start = START + (week * datetime.timedelta(7))
week_end = START + ((week + 1) * datetime.timedelta(7))
good = []
lame = []
skip = []
userlist = []
class User(object):
pass
for (un, rec) in users.items():
u = User()
u.username = un
u.links = rec['links']
u.start = rec['start']
u.end = rec.get('end')
u.skip = parse_skip(rec)
u.weeks = report.get(un, [])
userlist.append(u)
def user_key(u):
return (u.start, u.username)
userlist.sort(key=user_key)
for u in userlist:
user_start = parse(u.start, default=START)
if u.end and parse(u.end, default=START) <= week_start:
continue
if should_skip(u.skip, week):
pass
elif user_start > week_start:
skip.append(u)
elif len(u.weeks) <= week or not u.weeks[week]:
lame.append(u)
else:
good.append(u)
debts = get_debts()
return Template(filename=path, output_encoding='utf-8').render(
week=week, week_start=week_start,week_end=week_end,
good=good, lame=lame, skip=skip, userlist=userlist,
pool=get_balance('Pool'), paid=get_balance('Pool:Paid'),
debts=debts, **kwargs)
开发者ID:nelhage,项目名称:iron-blogger,代码行数:60,代码来源:render.py
示例8: settings
def settings(self):
core_path = os.path.join(get_root_path('indico'), 'modules', 'events', 'themes.yaml')
with open(core_path) as f:
core_data = f.read()
core_settings = yaml.safe_load(core_data)
# YAML doesn't give us access to anchors so we need to include the base yaml.
# Since duplicate keys are invalid (and may start failing in the future) we
# rename them - this also makes it easy to throw them away after parsing the
# file provided by a plugin.
core_data = re.sub(r'^(\S+:)$', r'__core_\1', core_data, flags=re.MULTILINE)
for plugin, path in values_from_signal(signals.plugin.get_event_themes_files.send(), return_plugins=True):
with open(path) as f:
data = f.read()
settings = {k: v
for k, v in yaml.safe_load(core_data + '\n' + data).viewitems()
if not k.startswith('__core_')}
# We assume there's no more than one theme plugin that provides defaults.
# If that's not the case the last one "wins". We could reject this but it
# is quite unlikely that people have multiple theme plugins in the first
# place, even more so theme plugins that specify defaults.
core_settings['defaults'].update(settings.get('defaults', {}))
# Same for definitions - we assume plugin authors are responsible enough
# to avoid using definition names that are likely to cause collisions.
# Either way, if someone does this on purpose chances are good they want
# to override a default style so let them do so...
for name, definition in settings.get('definitions', {}).viewitems():
definition['plugin'] = plugin
definition.setdefault('user_visible', False)
core_settings['definitions'][name] = definition
return core_settings
开发者ID:bkolobara,项目名称:indico,代码行数:30,代码来源:settings.py
示例9: test_for_consistency
def test_for_consistency(tests_path):
"""
Ensure that there is consistency between environment.yml dependencies
and conda.recipe/meta.yaml requirements.
"""
dev_pkgs = set([
'pytest',
'pytest-pep8',
'pytest-xdist',
'pycodestyle',
'pylint',
'coverage'
])
# read conda.recipe/meta.yaml requirements
meta_file = os.path.join(tests_path, '..', '..',
'conda.recipe', 'meta.yaml')
with open(meta_file, 'r') as stream:
meta = yaml.safe_load(stream)
bld = set(meta['requirements']['build'])
run = set(meta['requirements']['run'])
# confirm conda.recipe/meta.yaml build and run requirements are the same
assert bld == run
# read environment.yml dependencies
envr_file = os.path.join(tests_path, '..', '..',
'environment.yml')
with open(envr_file, 'r') as stream:
envr = yaml.safe_load(stream)
env = set(envr['dependencies'])
# confirm that extras in env (relative to run) equal the dev_pkgs set
extras = env - run
assert extras == dev_pkgs
开发者ID:open-source-economics,项目名称:Tax-Calculator,代码行数:31,代码来源:test_4package.py
示例10: generate_confs
def generate_confs(tileset, ignore_warnings=True, renderd=False):
"""
Takes a Tileset object and returns mapproxy and seed config files
"""
# Start with a sane configuration using MapProxy's defaults
mapproxy_config = load_default_config()
tileset_conf_json = get_mapproxy_conf(tileset)
tileset_conf = yaml.safe_load(tileset_conf_json)
# merge our config
load_config(mapproxy_config, config_dict=tileset_conf)
seed_conf_json = get_seed_conf(tileset)
seed_conf = yaml.safe_load(seed_conf_json)
errors, informal_only = validate_options(mapproxy_config)
if not informal_only or (errors and not ignore_warnings):
raise ConfigurationError('invalid configuration - {}'.format(', '.join(errors)))
mapproxy_cf = ProxyConfiguration(mapproxy_config, seed=seed, renderd=renderd)
errors, informal_only = validate_seed_conf(seed_conf)
if not informal_only:
raise SeedConfigurationError('invalid seed configuration - {}'.format(', '.join(errors)))
seed_cf = SeedingConfiguration(seed_conf, mapproxy_conf=mapproxy_cf)
return mapproxy_cf, seed_cf
开发者ID:JorgeMartinezG,项目名称:django-mapproxy,代码行数:28,代码来源:helpers.py
示例11: configure_sources
def configure_sources(update=False,
sources_var='install_sources',
keys_var='install_keys'):
"""
Configure multiple sources from charm configuration
Example config:
install_sources:
- "ppa:foo"
- "http://example.com/repo precise main"
install_keys:
- null
- "a1b2c3d4"
Note that 'null' (a.k.a. None) should not be quoted.
"""
sources = safe_load(config(sources_var))
keys = config(keys_var)
if keys is not None:
keys = safe_load(keys)
if isinstance(sources, basestring) and (
keys is None or isinstance(keys, basestring)):
add_source(sources, keys)
else:
if not len(sources) == len(keys):
msg = 'Install sources and keys lists are different lengths'
raise SourceConfigError(msg)
for src_num in range(len(sources)):
add_source(sources[src_num], keys[src_num])
if update:
apt_update(fatal=True)
开发者ID:CiscoSystems,项目名称:jujucharm-n1k,代码行数:31,代码来源:__init__.py
示例12: test_config_yaml
def test_config_yaml(self):
import yaml
cli.config(Args(), ["--format=yaml"])
data1 = yaml.safe_load(self.stdout.getvalue().strip().replace(helper.ROOT_PATH, "<ROOT>"))
data2 = yaml.safe_load(config_file("data.json"))
self.assertEqual(data1, data2)
开发者ID:silas,项目名称:rock,代码行数:7,代码来源:test_cli.py
示例13: get_playbook
def get_playbook(self):
"""
If necessary, fetch and read the playbook file
"""
playbook = self.config['playbook']
if isinstance(playbook, list):
# Multiple plays in a list
self.playbook = playbook
elif isinstance(playbook, str) and playbook.startswith(('http://',
'https://')):
response = requests.get(playbook)
response.raise_for_status()
self.playbook = yaml.safe_load(response.text)
elif isinstance(playbook, str):
try:
playbook_path = os.path.expanduser(playbook)
self.playbook_file = file(playbook_path)
playbook_yaml = yaml.safe_load(self.playbook_file)
self.playbook = playbook_yaml
except Exception:
log.error("Unable to read playbook file %s", playbook)
raise
else:
raise TypeError(
"playbook value must either be a list, URL or a filename")
log.info("Playbook: %s", self.playbook)
开发者ID:kawaguchi-s,项目名称:teuthology,代码行数:26,代码来源:ansible.py
示例14: __init__
def __init__(self, connections_amount=2, connection_propability=0.2):
self.__fire = 1000
with open("items.yaml") as data_file:
items_data = safe_load(data_file)
self.items = [ Item(elem['name'], elem['description'], elem['capacity'])
for elem in items_data ]
with open("rooms.yaml") as data_file:
rooms_data = safe_load(data_file)
self.rooms = [ Room(
elem['name'],
elem['description'],
random.sample(self.items, random.randint(0,2)),
bool(elem.get('water_source', 0))
) for elem in rooms_data ]
self.graph = graph_gen(len(self.rooms), connections_amount,\
connection_propability)
random.shuffle(self.rooms)
for no, room in enumerate(self.rooms):
room.neighbors = [self.rooms[index].name for index in \
self.graph.neighbors(no)]
#we start at 0
self.fired_room = self.rooms[random.randint(1, len(self.rooms))].name
self.spread_fire()
开发者ID:Nozdi,项目名称:mud,代码行数:27,代码来源:server.py
示例15: validate_config
def validate_config(self, template_data, yaml_file):
try:
yaml.safe_load(template_data)
except (yaml.scanner.ScannerError, yaml.YAMLError) as e:
self.log.error("Config for file {} contains invalid yaml, got "
"error {}".format(yaml_file, e))
raise e
开发者ID:openstack,项目名称:tripleo-common,代码行数:7,代码来源:config.py
示例16: test_save_shortcut_updated
def test_save_shortcut_updated(self):
OLD_YML = """\
projects:
default: 12345
prod: 33333
"""
runner = CliRunner()
with runner.isolated_filesystem():
with open('scrapinghub.yml', 'w') as f:
f.write(textwrap.dedent(OLD_YML))
conf = ShubConfig()
conf.load_file('scrapinghub.yml')
del conf.projects['prod']
print(conf.projects)
conf.save('scrapinghub.yml')
with open('scrapinghub.yml', 'r') as f:
new_yml = yaml.safe_load(f)
# Should not contain 'projects'
self.assertEqual(new_yml, {'project': 12345})
conf = ShubConfig()
conf.load_file('scrapinghub.yml')
# Should also work in reverse
conf.projects['prod'] = 33333
conf.save('scrapinghub.yml')
with open('scrapinghub.yml', 'r') as f:
new_yml = yaml.safe_load(f)
# Should not contain 'project' singleton
self.assertEqual(
new_yml,
{'projects': {'default': 12345, 'prod': 33333}},
)
# Make sure it is readable again
ShubConfig().load_file('scrapinghub.yml')
开发者ID:scrapinghub,项目名称:shub,代码行数:35,代码来源:test_config.py
示例17: get_disks
def get_disks(vm_):
"""
Return the disks of a named vm
CLI Example::
salt '*' virt.get_disks <vm name>
"""
disks = {}
doc = minidom.parse(StringIO.StringIO(get_xml(vm_)))
for elem in doc.getElementsByTagName("disk"):
sources = elem.getElementsByTagName("source")
targets = elem.getElementsByTagName("target")
if len(sources) > 0:
source = sources[0]
else:
continue
if len(targets) > 0:
target = targets[0]
else:
continue
if "dev" in target.attributes.keys() and "file" in source.attributes.keys():
disks[target.getAttribute("dev")] = {"file": source.getAttribute("file")}
for dev in disks:
try:
disks[dev].update(
yaml.safe_load(
subprocess.Popen(
"qemu-img info " + disks[dev]["file"], shell=True, stdout=subprocess.PIPE
).communicate()[0]
)
)
except TypeError:
disks[dev].update(yaml.safe_load("image: Does not exist"))
return disks
开发者ID:ralexandru,项目名称:salt,代码行数:35,代码来源:virt.py
示例18: generate_ui
def generate_ui(path, modules=[]):
"""Takes a path to a YAML UI definition, and generates a UI tree for it.
:param definition: A UI definition representing the UI to be created.
:param modules: (Optional) A list of module names which need to be
imported in order to generate the UI tree. This should include all
module names which define custom widgets or callbacks using
decorators that are used in the definition.
"""
for module in modules:
importlib.import_module(module)
with open(path, 'r') as ui_file:
ui = yaml.safe_load(ui_file)
full_style = {}
for style_path in ui.get('include', []):
with open(style_path, 'r') as style_file:
style = yaml.safe_load(style_file)
for definition in style:
full_style[definition['name']] = definition['properties']
definition = ui['definition']
root_class = yamlui.class_mapping.get(definition['object'])
if root_class is None:
raise Exception('ERROR: Root class is an unrecognised widget type.')
root = root_class(definition, style=full_style)
ui_name = os.path.basename(path)
yamlui.trees[ui_name] = build_dictionary(root, ui_name)
return root
开发者ID:ColdrickSotK,项目名称:yamlui,代码行数:33,代码来源:parsing.py
示例19: wait_until_started
def wait_until_started(self, wait_load=True):
""" Wait until server is started.
Server consists of two parts:
1) wait until server is listening on sockets
2) wait until server tells us his status
"""
if wait_load:
msg = 'entering the event loop|will retry binding|hot standby mode'
p = self.process if not self.gdb and not self.lldb else None
self.logfile_pos.seek_wait(msg, p, self.name)
while True:
try:
temp = AdminConnection('localhost', self.admin.port)
if not wait_load:
ans = yaml.safe_load(temp.execute("2 + 2"))
return True
ans = yaml.safe_load(temp.execute('box.info.status'))[0]
if ans in ('running', 'hot_standby', 'orphan'):
return True
elif ans in ('loading'):
continue
else:
raise Exception(
"Strange output for `box.info.status`: %s" % (ans)
)
except socket.error as e:
if e.errno == errno.ECONNREFUSED:
time.sleep(0.1)
continue
raise
开发者ID:tarantool,项目名称:test-run,代码行数:32,代码来源:tarantool_server.py
示例20: main
def main():
parser = argparse.ArgumentParser()
parser.add_argument('--old', required=True,
help='Current policy file')
parser.add_argument('--new', required=True,
help='New policy file')
parser.add_argument('--mode',
choices=['add', 'remove'],
default='remove',
help='Diffs to be shown')
parsed_args = parser.parse_args()
with open(parsed_args.old) as f:
old_data = yaml.safe_load(f)
with open(parsed_args.new) as f:
new_data = yaml.safe_load(f)
added = set(new_data.keys()) - set(old_data.keys())
removed = set(old_data.keys()) - set(new_data.keys())
if parsed_args.mode == 'remove':
for key in sorted(removed):
print(key)
if parsed_args.mode == 'add':
for key in sorted(added):
print(key)
开发者ID:CCI-MOC,项目名称:horizon,代码行数:28,代码来源:policy-diff.py
注:本文中的yaml.safe_load函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台,相关代码片段筛选自各路编程大神贡献的开源项目,源码版权归原作者所有,传播和使用请参考对应项目的License;未经允许,请勿转载。 |
请发表评论