Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
def _exception(name, e, always_dump=False):
print()
if pseudo_host.isset():
sys.stderr.write('--> [{0}]: {1}: '.format(
colored(pseudo_host.name, attrs=['bold']),
colored(name, 'red', attrs=['bold'])
))
else:
sys.stderr.write('--> {0}: '.format(colored(name, 'red', attrs=['bold'])))
if e:
logger.warning(e)
if arguments.get('debug') or always_dump:
dump_trace(sys.exc_info())
_exit(1)
def show_warning():
logger.warning('The @ansible connector is in Alpha!')
current_op_hash = pseudo_state.current_op_hash
current_op_meta = pseudo_state.op_meta.get(current_op_hash)
if current_op_meta:
op_name = ', '.join(current_op_meta['names'])
sys.stderr.write('--> {0}{1}{2}: '.format(
pseudo_host.print_prefix,
click.style(name, 'red', bold=True),
' (operation={0})'.format(op_name) if op_name else '',
))
else:
sys.stderr.write(
'--> {0}: '.format(click.style(name, 'red', bold=True)),
)
logger.warning(self)
def hosts(self, hosts):
logger.warning((
'Use of `State.hosts` is deprecated, '
'please use normal `if` statements instead.'
))
hosts = ensure_host_list(hosts, inventory=self.inventory)
# Store the previous value
old_limit_hosts = self.limit_hosts
# Limit the new hosts to a subset of the old hosts if they existed
if old_limit_hosts is not None:
hosts = [
host for host in hosts
if host in old_limit_hosts
]
+ recursive_submodules: update git submodules recursively
+ [DEPRECATED] use_ssh_user: whether to use the SSH user to clone/pull
SSH user (deprecated, please use ``preserve_sudo_env``):
This is an old hack from pyinfra <0.4 which did not support the global
kwarg ``preserve_sudo_env``. It does the following:
* makes the target directory writeable by all
* clones/pulls w/o sudo as the connecting SSH user
* removes other/group write permissions - unless group is defined, in
which case only other
'''
if use_ssh_user:
logger.warning(
'Use of `use_ssh_user` is deprecated, please use `preserve_sudo_env` instead.',
)
# Ensure our target directory exists
yield files.directory(state, host, target)
# If we're going to chown this after clone/pull, and we're sudo'd, we need to make the
# directory writeable by the SSH user
if use_ssh_user:
yield chmod(target, 'go+w', recursive=True)
# Do we need to scan for the remote host key?
if ssh_keyscan:
# Attempt to parse the domain from the git repository
domain = re.match(r'^[a-zA-Z0-9]+@([0-9a-zA-Z\.\-]+)', source)
def when(self, predicate):
logger.warning((
'Use of `State.when` is deprecated, '
'please use normal `if` statements instead.'
))
# Truth-y? Just yield/end, nothing happens here!
if predicate:
yield
return
# Otherwise limit any operations within to match no hosts
with self.hosts([]):
yield
paths = [p]
while os.path.islink(p):
p = os.path.normcase(os.path.join(os.path.dirname(p), os.readlink(p)))
paths.append(p)
# In Cygwin paths like 'c:\...' and '\cygdrive\c\...' are possible
if p_venv.startswith('\\cygdrive'):
p_venv = p_venv[11:]
elif len(p_venv) >= 2 and p_venv[1] == ':':
p_venv = p_venv[2:]
if any(p_venv in p for p in paths):
# Running properly in the virtualenv, don't need to do anything
return
logger.warning((
'Attempting to work in a virtualenv.\n'
' If you encounter problems, please install pyinfra inside the virtualenv.'
))
print()
if sys.platform == 'win32':
virtual_env = os.path.join(
os.environ['VIRTUAL_ENV'],
'Lib',
'site-packages',
)
else:
virtual_env = os.path.join(
os.environ['VIRTUAL_ENV'],
'lib',
'python%d.%d' % sys.version_info[:2],
def add_limited_op(state, op_func, hosts, *args, **kwargs):
'''
DEPRECATED: please use ``add_op`` with the ``hosts`` kwarg.
'''
# COMPAT w/ <0.4
# TODO: remove this function
logger.warning((
'Use of `add_limited_op` is deprecated, '
'please use `add_op` with the `hosts` kwarg instead.'
))
if not isinstance(hosts, (list, tuple)):
hosts = [hosts]
# Set the limit
state.limit_hosts = hosts
# Add the op
add_op(state, op_func, *args, **kwargs)
# Remove the limit
state.limit_hosts = []
def __getattr__(self, key):
'''
DEPRECATED: please use ``Inventory.get_group`` instead.
'''
# COMPAT w/ <0.4
# TODO: remove this function
logger.warning((
'Use of Inventory. is deprecated, '
'please use `Inventory.get_group` instead.'
))
if key in self.groups:
return self.groups[key]
raise NoGroupError('No such group: {0}'.format(key))