本文整理汇总了Python中salt._compat.urlparse函数的典型用法代码示例。如果您正苦于以下问题：Python urlparse函数的具体用法？Python urlparse怎么用？Python urlparse使用的例子？那么恭喜您, 这里精选的函数代码示例或许可以为您提供帮助。
在下文中一共展示了urlparse函数的13个代码示例，这些例子默认根据受欢迎程度排序。您可以为喜欢或者感觉有用的代码点赞，您的评价将有助于我们的系统推荐出更棒的Python代码示例。
示例1: source_list
def source_list(source, source_hash, env):
'''
Check the source list and return the source to use
'''
if isinstance(source, list):
# get the master file list
mfiles = __salt__['cp.list_master'](env)
mdirs = __salt__['cp.list_master_dirs'](env)
for single in source:
if isinstance(single, dict):
# check the proto, if it is http or ftp then download the file
# to check, if it is salt then check the master list
if len(single) != 1:
continue
single_src = next(iter(single))
single_hash = single[single_src]
proto = urlparse(single_src).scheme
if proto == 'salt':
if single_src in mfiles:
source = single_src
break
elif proto.startswith('http') or proto == 'ftp':
dest = salt.utils.mkstemp()
fn_ = __salt__['cp.get_url'](single_src, dest)
os.remove(fn_)
if fn_:
source = single_src
source_hash = single_hash
break
elif isinstance(single, string_types):
if single[7:] in mfiles or single[7:] in mdirs:
source = single
break
return source, source_hash
开发者ID:inthecloud247，项目名称:salt，代码行数:34，代码来源:file.py
示例2: is_cached
def is_cached(self, path, saltenv='base', env=None):
'''
Returns the full path to a file if it is cached locally on the minion
otherwise returns a blank string
'''
if env is not None:
salt.utils.warn_until(
'Boron',
'Passing a salt environment should be done using \'saltenv\' '
'not \'env\'. This functionality will be removed in Salt '
'Boron.'
)
# Backwards compatibility
saltenv = env
# support url
url_data = urlparse(path)
localsfilesdest = os.path.join(
self.opts['cachedir'], 'localfiles', path.lstrip('/'))
filesdest = os.path.join(
self.opts['cachedir'], 'files', saltenv, path.lstrip('salt://'))
# support url
urlcacheddest = salt.utils.path_join(
self.opts['cachedir'], 'extrn_files', saltenv, url_data.netloc, url_data.path)
if os.path.exists(filesdest):
return filesdest
elif os.path.exists(localsfilesdest):
return localsfilesdest
elif os.path.exists(urlcacheddest):
return urlcacheddest
return ''
开发者ID:MadeiraCloud，项目名称:salt，代码行数:34，代码来源:fileclient.py
示例3: get_template
def get_template(self, url, dest, template="jinja", makedirs=False, env="base", **kwargs):
"""
Cache a file then process it as a template
"""
kwargs["env"] = env
url_data = urlparse(url)
sfn = self.cache_file(url, env)
if not os.path.exists(sfn):
return ""
if template in salt.utils.templates.TEMPLATE_REGISTRY:
data = salt.utils.templates.TEMPLATE_REGISTRY[template](sfn, **kwargs)
else:
log.error("Attempted to render template with unavailable engine " "{0}".format(template))
salt.utils.safe_rm(data["data"])
return ""
if not data["result"]:
# Failed to render the template
log.error("Failed to render template with error: {0}".format(data["data"]))
return ""
if not dest:
# No destination passed, set the dest as an extrn_files cache
dest = salt.utils.path_join(self.opts["cachedir"], "extrn_files", env, url_data.netloc, url_data.path)
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
if makedirs:
os.makedirs(destdir)
else:
salt.utils.safe_rm(data["data"])
return ""
shutil.move(data["data"], dest)
return dest
开发者ID:Barrybaby，项目名称:salt，代码行数:31，代码来源:fileclient.py
示例4: get_url
def get_url(self, url, dest, makedirs=False, env="base"):
"""
Get a single file from a URL.
"""
url_data = urlparse(url)
if url_data.scheme == "salt":
return self.get_file(url, dest, makedirs, env)
if dest:
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
if makedirs:
os.makedirs(destdir)
else:
return ""
else:
dest = salt.utils.path_join(self.opts["cachedir"], "extrn_files", env, url_data.netloc, url_data.path)
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
os.makedirs(destdir)
try:
with contextlib.closing(url_open(url)) as srcfp:
with salt.utils.fopen(dest, "wb") as destfp:
shutil.copyfileobj(srcfp, destfp)
return dest
except HTTPError as ex:
raise MinionError(
"HTTP error {0} reading {1}: {3}".format(
ex.code, url, *BaseHTTPServer.BaseHTTPRequestHandler.responses[ex.code]
)
)
except URLError as ex:
raise MinionError("Error reading {0}: {1}".format(url, ex.reason))
开发者ID:Barrybaby，项目名称:salt，代码行数:32，代码来源:fileclient.py
示例5: get_template
def get_template(
self,
url,
dest,
template='jinja',
makedirs=False,
env='base',
**kwargs):
'''
Cache a file then process it as a template
'''
kwargs['env'] = env
url_data = urlparse(url)
sfn = self.cache_file(url, env)
if not os.path.exists(sfn):
return ''
if template in salt.utils.templates.TEMPLATE_REGISTRY:
data = salt.utils.templates.TEMPLATE_REGISTRY[template](
sfn,
**kwargs
)
else:
log.error('Attempted to render template with unavailable engine '
'{0}'.format(template))
return ''
if not data['result']:
# Failed to render the template
log.error(
'Failed to render template with error: {0}'.format(
data['data']
)
)
return ''
if not dest:
# No destination passed, set the dest as an extrn_files cache
dest = salt.utils.path_join(
self.opts['cachedir'],
'extrn_files',
env,
url_data.netloc,
url_data.path
)
# If Salt generated the dest name, create any required dirs
makedirs = True
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
if makedirs:
os.makedirs(destdir)
else:
salt.utils.safe_rm(data['data'])
return ''
shutil.move(data['data'], dest)
return dest
开发者ID:victorywang80，项目名称:Maintenance，代码行数:54，代码来源:fileclient.py
示例6: get_url
def get_url(self, url, dest, makedirs=False, env='base'):
'''
Get a single file from a URL.
'''
url_data = urlparse(url)
if url_data.scheme == 'salt':
return self.get_file(url, dest, makedirs, env)
if dest:
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
if makedirs:
os.makedirs(destdir)
else:
return ''
else:
dest = salt.utils.path_join(
self.opts['cachedir'],
'extrn_files',
env,
url_data.netloc,
url_data.path
)
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
os.makedirs(destdir)
if url_data.username is not None \
and url_data.scheme in ('http', 'https'):
_, netloc = url_data.netloc.split('@', 1)
fixed_url = urlunparse(
(url_data.scheme, netloc, url_data.path,
url_data.params, url_data.query, url_data.fragment))
passwd_mgr = url_passwd_mgr()
passwd_mgr.add_password(
None, fixed_url, url_data.username, url_data.password)
auth_handler = url_auth_handler(passwd_mgr)
opener = url_build_opener(auth_handler)
url_install_opener(opener)
else:
fixed_url = url
try:
with contextlib.closing(url_open(fixed_url)) as srcfp:
with salt.utils.fopen(dest, 'wb') as destfp:
shutil.copyfileobj(srcfp, destfp)
return dest
except HTTPError as ex:
raise MinionError('HTTP error {0} reading {1}: {3}'.format(
ex.code,
url,
*BaseHTTPServer.BaseHTTPRequestHandler.responses[ex.code]))
except URLError as ex:
raise MinionError('Error reading {0}: {1}'.format(url, ex.reason))
开发者ID:victorywang80，项目名称:Maintenance，代码行数:51，代码来源:fileclient.py
示例7: get_url
def get_url(self, url, dest, makedirs=False, env='base'):
'''
Get a single file from a URL.
'''
url_data = urlparse(url)
if url_data.scheme == 'salt':
return self.get_file(url, dest, makedirs, env)
if dest:
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
if makedirs:
os.makedirs(destdir)
else:
return ''
else:
dest = os.path.normpath(
os.sep.join([
self.opts['cachedir'],
'extrn_files',
env,
url_data.netloc,
url_data.path]))
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
os.makedirs(destdir)
try:
with contextlib.closing(url_open(url)) as srcfp:
with open(dest, 'wb') as destfp:
shutil.copyfileobj(srcfp, destfp)
return dest
except HTTPError as ex:
raise MinionError('HTTP error {0} reading {1}: {3}'.format(
ex.code,
url,
*BaseHTTPServer.BaseHTTPRequestHandler.responses[ex.code]))
except URLError as ex:
raise MinionError('Error reading {0}: {1}'.format(url, ex.reason))
return ''
开发者ID:twinshadow，项目名称:salt，代码行数:38，代码来源:fileclient.py
示例8: get_managed
def get_managed(
name,
template,
source,
source_hash,
user,
group,
mode,
env,
context,
defaults,
**kwargs):
'''
Return the managed file data for file.managed
'''
# If the file is a template and the contents is managed
# then make sure to copy it down and templatize things.
sfn = ''
source_sum = {}
if template and source:
sfn = __salt__['cp.cache_file'](source, env)
if not os.path.exists(sfn):
return sfn, {}, 'File "{0}" could not be found'.format(sfn)
if template in salt.utils.templates.TEMPLATE_REGISTRY:
context_dict = defaults if defaults else {}
if context:
context_dict.update(context)
data = salt.utils.templates.TEMPLATE_REGISTRY[template](
sfn,
name=name,
source=source,
user=user,
group=group,
mode=mode,
env=env,
context=context_dict,
salt=__salt__,
pillar=__pillar__,
grains=__grains__,
opts=__opts__,
**kwargs
)
else:
return sfn, {}, ('Specified template format {0} is not supported'
).format(template)
if data['result']:
sfn = data['data']
hsum = get_hash(sfn)
source_sum = {'hash_type': 'md5',
'hsum': hsum}
else:
__clean_tmp(sfn)
return sfn, {}, data['data']
else:
# Copy the file down if there is a source
if source:
if urlparse(source).scheme == 'salt':
source_sum = __salt__['cp.hash_file'](source, env)
if not source_sum:
return '', {}, 'Source file {0} not found'.format(source)
elif source_hash:
protos = ['salt', 'http', 'ftp']
if urlparse(source_hash).scheme in protos:
# The source_hash is a file on a server
hash_fn = __salt__['cp.cache_file'](source_hash)
if not hash_fn:
return '', {}, 'Source hash file {0} not found'.format(
source_hash)
hash_fn_fopen = salt.utils.fopen(hash_fn, 'r')
for line in hash_fn_fopen.read().splitlines():
line = line.strip()
if ' ' not in line:
hashstr = line
break
elif line.startswith('{0} '.format(name)):
hashstr = line.split()[1]
break
else:
hashstr = '' # NOT FOUND
comps = hashstr.split('=')
if len(comps) < 2:
return '', {}, ('Source hash file {0} contains an '
'invalid hash format, it must be in '
'the format <hash type>=<hash>'
).format(source_hash)
source_sum['hsum'] = comps[1].strip()
source_sum['hash_type'] = comps[0].strip()
else:
# The source_hash is a hash string
comps = source_hash.split('=')
if len(comps) < 2:
return '', {}, ('Source hash file {0} contains an '
'invalid hash format, it must be in '
'the format <hash type>=<hash>'
).format(source_hash)
source_sum['hsum'] = comps[1].strip()
source_sum['hash_type'] = comps[0].strip()
else:
return '', {}, ('Unable to determine upstream hash of'
#.........这里部分代码省略.........
开发者ID:inthecloud247，项目名称:salt，代码行数:101，代码来源:file.py
示例9: manage_file
def manage_file(name,
sfn,
ret,
source,
source_sum,
user,
group,
mode,
env,
backup):
'''
Checks the destination against what was retrieved with get_managed and
makes the appropriate modifications (if necessary).
'''
if not ret:
ret = {'name': name,
'changes': {},
'comment': '',
'result': True}
# Check changes if the target file exists
if os.path.isfile(name):
# Only test the checksums on files with managed contents
if source:
name_sum = get_hash(name, source_sum['hash_type'])
# Check if file needs to be replaced
if source and source_sum['hsum'] != name_sum:
if not sfn:
sfn = __salt__['cp.cache_file'](source, env)
if not sfn:
return _error(
ret, 'Source file {0} not found'.format(source))
# If the downloaded file came from a non salt server source verify
# that it matches the intended sum value
if urlparse(source).scheme != 'salt':
dl_sum = get_hash(sfn, source_sum['hash_type'])
if dl_sum != source_sum['hsum']:
ret['comment'] = ('File sum set for file {0} of {1} does '
'not match real sum of {2}'
).format(
name,
source_sum['hsum'],
dl_sum
)
ret['result'] = False
return ret
# Check to see if the files are bins
if _is_bin(sfn) or _is_bin(name):
ret['changes']['diff'] = 'Replace binary file'
else:
with nested(salt.utils.fopen(sfn, 'rb'),
salt.utils.fopen(name, 'rb')) as (src, name_):
slines = src.readlines()
nlines = name_.readlines()
# Print a diff equivalent to diff -u old new
ret['changes']['diff'] = (''.join(difflib
.unified_diff(nlines,
slines)))
# Pre requisites are met, and the file needs to be replaced, do it
try:
salt.utils.copyfile(
sfn,
name,
__salt__['config.backup_mode'](backup),
__opts__['cachedir'])
except IOError:
__clean_tmp(sfn)
return _error(
ret, 'Failed to commit change, permission error')
ret, perms = check_perms(name, ret, user, group, mode)
if ret['changes']:
ret['comment'] = 'File {0} updated'.format(name)
elif not ret['changes'] and ret['result']:
ret['comment'] = 'File {0} is in the correct state'.format(name)
__clean_tmp(sfn)
return ret
else:
# Only set the diff if the file contents is managed
if source:
# It is a new file, set the diff accordingly
ret['changes']['diff'] = 'New file'
# Apply the new file
if not sfn:
sfn = __salt__['cp.cache_file'](source, env)
if not sfn:
return ret.error(
ret, 'Source file {0} not found'.format(source))
# If the downloaded file came from a non salt server source verify
# that it matches the intended sum value
if urlparse(source).scheme != 'salt':
dl_sum = get_hash(name, source_sum['hash_type'])
if dl_sum != source_sum['hsum']:
ret['comment'] = ('File sum set for file {0} of {1} does '
'not match real sum of {2}'
).format(
name,
#.........这里部分代码省略.........
开发者ID:inthecloud247，项目名称:salt，代码行数:101，代码来源:file.py
示例10: get_template
def get_template(
self,
url,
dest,
template='jinja',
makedirs=False,
saltenv='base',
env=None,
**kwargs):
'''
Cache a file then process it as a template
'''
if env is not None:
salt.utils.warn_until(
'Boron',
'Passing a salt environment should be done using \'saltenv\' '
'not \'env\'. This functionality will be removed in Salt '
'Boron.'
)
# Backwards compatibility
saltenv = env
kwargs['saltenv'] = saltenv
url_data = urlparse(url)
sfn = self.cache_file(url, saltenv)
if not os.path.exists(sfn):
return ''
if template in salt.utils.templates.TEMPLATE_REGISTRY:
data = salt.utils.templates.TEMPLATE_REGISTRY[template](
sfn,
**kwargs
)
else:
log.error('Attempted to render template with unavailable engine '
'{0}'.format(template))
return ''
if not data['result']:
# Failed to render the template
log.error(
'Failed to render template with error: {0}'.format(
data['data']
)
)
return ''
if not dest:
# No destination passed, set the dest as an extrn_files cache
dest = salt.utils.path_join(
self.opts['cachedir'],
'extrn_files',
saltenv,
url_data.netloc,
url_data.path
)
# If Salt generated the dest name, create any required dirs
makedirs = True
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
if makedirs:
os.makedirs(destdir)
else:
salt.utils.safe_rm(data['data'])
return ''
shutil.move(data['data'], dest)
return dest
开发者ID:AccelerationNet，项目名称:salt，代码行数:65，代码来源:fileclient.py
示例11: get_url
def get_url(self, url, dest, makedirs=False, saltenv='base', env=None):
'''
Get a single file from a URL.
'''
if env is not None:
salt.utils.warn_until(
'Boron',
'Passing a salt environment should be done using \'saltenv\' '
'not \'env\'. This functionality will be removed in Salt '
'Boron.'
)
# Backwards compatibility
saltenv = env
url_data = urlparse(url)
if url_data.scheme == 'salt':
return self.get_file(url, dest, makedirs, saltenv)
if dest:
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
if makedirs:
os.makedirs(destdir)
else:
return ''
else:
dest = salt.utils.path_join(
self.opts['cachedir'],
'extrn_files',
saltenv,
url_data.netloc,
url_data.path
)
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
os.makedirs(destdir)
if url_data.scheme == 's3':
try:
salt.utils.s3.query(method='GET',
bucket=url_data.netloc,
path=url_data.path[1:],
return_bin=False,
local_file=dest,
action=None,
key=self.opts.get('s3.key', None),
keyid=self.opts.get('s3.keyid', None),
service_url=self.opts.get('s3.service_url',
None),
verify_ssl=self.opts.get('s3.verify_ssl',
True))
return dest
except Exception as ex:
raise MinionError('Could not fetch from {0}'.format(url))
if url_data.scheme == 'swift':
try:
swift_conn = SaltSwift(self.opts.get('keystone.user', None),
self.opts.get('keystone.tenant', None),
self.opts.get('keystone.auth_url', None),
self.opts.get('keystone.password', None))
swift_conn.get_object(url_data.netloc,
url_data.path[1:],
dest)
return dest
except Exception as ex:
raise MinionError('Could not fetch from {0}'.format(url))
if url_data.username is not None \
and url_data.scheme in ('http', 'https'):
_, netloc = url_data.netloc.split('@', 1)
fixed_url = urlunparse(
(url_data.scheme, netloc, url_data.path,
url_data.params, url_data.query, url_data.fragment))
passwd_mgr = url_passwd_mgr()
passwd_mgr.add_password(
None, fixed_url, url_data.username, url_data.password)
auth_handler = url_auth_handler(passwd_mgr)
opener = url_build_opener(auth_handler)
url_install_opener(opener)
else:
fixed_url = url
try:
req = requests.get(fixed_url)
with salt.utils.fopen(dest, 'wb') as destfp:
destfp.write(req.content)
return dest
except HTTPError as ex:
raise MinionError('HTTP error {0} reading {1}: {3}'.format(
ex.code,
url,
*BaseHTTPServer.BaseHTTPRequestHandler.responses[ex.code]))
except URLError as ex:
raise MinionError('Error reading {0}: {1}'.format(url, ex.reason))
开发者ID:AccelerationNet，项目名称:salt，代码行数:93，代码来源:fileclient.py
示例12: get_url
def get_url(self, url, dest, makedirs=False, saltenv='base', env=None):
'''
Get a single file from a URL.
'''
if env is not None:
salt.utils.warn_until(
'Boron',
'Passing a salt environment should be done using \'saltenv\' '
'not \'env\'. This functionality will be removed in Salt '
'Boron.'
)
# Backwards compatibility
saltenv = env
url_data = urlparse(url)
if url_data.scheme in ('file', ''):
# Local filesystem
if not os.path.isabs(url_data.path):
raise CommandExecutionError(
'Path {0!r} is not absolute'.format(url_data.path)
)
return url_data.path
if url_data.scheme == 'salt':
return self.get_file(url, dest, makedirs, saltenv)
if dest:
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
if makedirs:
os.makedirs(destdir)
else:
return ''
else:
if salt.utils.is_windows():
netloc = salt.utils.sanitize_win_path_string(url_data.netloc)
else:
netloc = url_data.netloc
dest = salt.utils.path_join(
self.opts['cachedir'],
'extrn_files',
saltenv,
netloc,
url_data.path
)
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
os.makedirs(destdir)
if url_data.scheme == 's3':
try:
salt.utils.s3.query(method='GET',
bucket=url_data.netloc,
path=url_data.path[1:],
return_bin=False,
local_file=dest,
action=None,
key=self.opts.get('s3.key', None),
keyid=self.opts.get('s3.keyid', None),
service_url=self.opts.get('s3.service_url',
None),
verify_ssl=self.opts.get('s3.verify_ssl',
True))
return dest
except Exception:
raise MinionError('Could not fetch from {0}'.format(url))
if url_data.scheme == 'swift':
try:
swift_conn = SaltSwift(self.opts.get('keystone.user', None),
self.opts.get('keystone.tenant', None),
self.opts.get('keystone.auth_url', None),
self.opts.get('keystone.password', None))
swift_conn.get_object(url_data.netloc,
url_data.path[1:],
dest)
return dest
except Exception:
raise MinionError('Could not fetch from {0}'.format(url))
get_kwargs = {}
if url_data.username is not None \
and url_data.scheme in ('http', 'https'):
_, netloc = url_data.netloc.split('@', 1)
fixed_url = urlunparse(
(url_data.scheme, netloc, url_data.path,
url_data.params, url_data.query, url_data.fragment))
get_kwargs['auth'] = (url_data.username, url_data.password)
else:
fixed_url = url
try:
if requests.__version__[0] == '0':
# 'stream' was called 'prefetch' before 1.0, with flipped meaning
get_kwargs['prefetch'] = False
else:
get_kwargs['stream'] = True
response = requests.get(fixed_url, **get_kwargs)
response.raise_for_status()
with salt.utils.fopen(dest, 'wb') as destfp:
for chunk in response.iter_content(chunk_size=32*1024):
#.........这里部分代码省略.........
开发者ID:Quarky9，项目名称:states，代码行数:101，代码来源:fileclient.py
示例13: get_url
def get_url(self, url, dest, makedirs=False, saltenv='base', env=None):
'''
Get a single file from a URL.
'''
if env is not None:
salt.utils.warn_until(
'Boron',
'Passing a salt environment should be done using \'saltenv\' '
'not \'env\'. This functionality will be removed in Salt '
'Boron.'
)
# Backwards compatibility
saltenv = env
url_data = urlparse(url)
if url_data.scheme == 'salt':
return self.get_file(url, dest, makedirs, saltenv)
if dest:
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
if makedirs:
os.makedirs(destdir)
else:
return ''
else:
dest = salt.utils.path_join(
self.opts['cachedir'],
'extrn_files',
saltenv,
url_data.netloc,
url_data.path
)
destdir = os.path.dirname(dest)
if not os.path.isdir(destdir):
os.makedirs(destdir)
if url_data.username is not None \
and url_data.scheme in ('http', 'https'):
_, netloc = url_data.netloc.split('@', 1)
fixed_url = urlunparse(
(url_data.scheme, netloc, url_data.path,
url_data.params, url_data.query, url_data.fragment))
passwd_mgr = url_passwd_mgr()
passwd_mgr.add_password(
None, fixed_url, url_data.username, url_data.password)
auth_handler = url_auth_handler(passwd_mgr)
opener = url_build_opener(auth_handler)
url_install_opener(opener)
else:
fixed_url = url
try:
req = requests.get(fixed_url)
with salt.utils.fopen(dest, 'wb') as destfp:
destfp.write(req.content)
return dest
except HTTPError as ex:
raise MinionError('HTTP error {0} reading {1}: {3}'.format(
ex.code,
url,
*BaseHTTPServer.BaseHTTPRequestHandler.responses[ex.code]))
except URLError as ex:
raise MinionError('Error reading {0}: {1}'.format(url, ex.reason))
开发者ID:MadeiraCloud，项目名称:salt，代码行数:61，代码来源:fileclient.py
注：本文中的salt._compat.urlparse函数示例由纯净天空整理自Github/MSDocs等源码及文档管理平台，相关代码片段筛选自各路编程大神贡献的开源项目，源码版权归原作者所有，传播和使用请参考对应项目的License；未经允许，请勿转载。 |
客服电话
APP下载
官方微信
请发表评论