我们从Python开源项目中，提取了以下13个代码示例，用于说明如何使用ansible.errors.AnsibleUndefinedVariable()。
def queue_exc(*arg, **kw): # pylint: disable=W0613 """Queue undefined variable exception""" _self = arg[0] if not isinstance(_self, AnsibleUndefinedVariable): # Run for AnsibleUndefinedVariable instance return _rslt_q = None for stack_trace in inspect.stack(): # Check if method to be skipped if stack_trace[3] in SKIP_METHODS: continue _frame = stack_trace[0] _locals = inspect.getargvalues(_frame).locals if 'self' not in _locals: continue # Check if current frame instance of worker if isinstance(_locals['self'], WorkerProcess): # Get queue to add exception _rslt_q = getattr(_locals['self'], '_rslt_q') if not _rslt_q: raise ValueError("No Queue found.") _rslt_q.put({"undefined_var": arg[3].message}, interceptor=True)
def is_template(self, data): ''' lets us know if data has a template''' if isinstance(data, string_types): try: new = self.do_template(data, fail_on_undefined=True) except (AnsibleUndefinedVariable, UndefinedError): return True except: return False return (new != data) elif isinstance(data, (list, tuple)): for v in data: if self.is_template(v): return True elif isinstance(data, dict): for k in data: if self.is_template(k) or self.is_template(data[k]): return True return False
def aspects(self): """Get aspects""" return { AnsibleUndefinedVariable: { r'__init__': dict( around_after=queue_exc )}, StrategyBase: { r'\brun\b': dict( before=extract_worker_exc(self.errors) # bind data store )}, TaskQueueManager: { r'\bsend_callback\b': dict(before=extract_setup_failures(self.errors)) } }
def _lookup(self, name, *args, **kwargs): instance = self._lookup_loader.get(name.lower(), loader=self._loader, templar=self) if instance is not None: wantlist = kwargs.pop('wantlist', False) from ansible.utils.listify import listify_lookup_plugin_terms loop_terms = listify_lookup_plugin_terms(terms=args, templar=self, loader=self._loader, fail_on_undefined=True, convert_bare=False) # safely catch run failures per #5059 try: ran = instance.run(loop_terms, variables=self._available_variables, **kwargs) except (AnsibleUndefinedVariable, UndefinedError) as e: raise AnsibleUndefinedVariable(e) except Exception as e: if self._fail_on_lookup_errors: raise ran = None if ran: from ansible.vars.unsafe_proxy import UnsafeProxy, wrap_var if wantlist: ran = wrap_var(ran) else: try: ran = UnsafeProxy(",".join(ran)) except TypeError: if isinstance(ran, list) and len(ran) == 1: ran = wrap_var(ran[0]) else: ran = wrap_var(ran) return ran else: raise AnsibleError("lookup plugin (%s) not found" % name)
def _lookup_variables(self, terms, variables): results = [] for x in terms: try: intermediate = listify_lookup_plugin_terms(x, templar=self._templar, loader=self._loader, fail_on_undefined=True) except UndefinedError as e: raise AnsibleUndefinedVariable("One of the nested variables was undefined. The error was: %s" % e) results.append(intermediate) return results
def run(self, terms, variables, **kwargs): if len(terms) <= 1: raise AnsibleUndefinedVariable("Mergedict takes 2 options") if not (isinstance(terms[0], dict) or isinstance(terms[0], list)): raise AnsibleUndefinedVariable("First option must be a dict or list") if not isinstance(terms[1], dict): raise AnsibleUndefinedVariable("Second option must be a dict") result=[] temp={} to_data = terms[0] from_dict = terms[1] for item in to_data: if isinstance(to_data, list): temp = self.dict_merge(from_dict.copy(), item) if isinstance(to_data, dict): if len(terms) == 3: if isinstance(terms[2], list): if item not in terms[2]: continue else: raise AnsibleUndefinedVariable("Third option must be a list") temp = self.dict_merge(from_dict.copy(), to_data[item]) temp['key'] = item result.append(temp) return result
def _post_validate_environment(self, attr, value, templar): ''' Override post validation of vars on the play, as we don't want to template these too early. ''' env = {} if value is not None: def _parse_env_kv(k, v): try: env[k] = templar.template(v, convert_bare=False) except AnsibleUndefinedVariable as e: if self.action in ('setup', 'gather_facts') and 'ansible_env' in to_native(e): # ignore as fact gathering sets ansible_env pass if isinstance(value, list): for env_item in value: if isinstance(env_item, dict): for k in env_item: _parse_env_kv(k, env_item[k]) else: isdict = templar.template(env_item, convert_bare=False) if isinstance(isdict, dict): env.update(isdict) else: display.warning("could not parse environment value, skipping: %s" % value) elif isinstance(value, dict): # should not really happen env = dict() for env_item in value: _parse_env_kv(env_item, value[env_item]) else: # at this point it should be a simple string, also should not happen env = templar.template(value, convert_bare=False) return env
def _lookup(self, name, *args, **kwargs): instance = self._lookup_loader.get(name.lower(), loader=self._loader, templar=self) if instance is not None: wantlist = kwargs.pop('wantlist', False) allow_unsafe = kwargs.pop('allow_unsafe', C.DEFAULT_ALLOW_UNSAFE_LOOKUPS) from ansible.utils.listify import listify_lookup_plugin_terms loop_terms = listify_lookup_plugin_terms(terms=args, templar=self, loader=self._loader, fail_on_undefined=True, convert_bare=False) # safely catch run failures per #5059 try: ran = instance.run(loop_terms, variables=self._available_variables, **kwargs) except (AnsibleUndefinedVariable, UndefinedError) as e: raise AnsibleUndefinedVariable(e) except Exception as e: if self._fail_on_lookup_errors: raise AnsibleError("An unhandled exception occurred while running the lookup plugin '%s'. Error was a %s, " "original message: %s" % (name, type(e), e)) ran = None if ran and not allow_unsafe: if wantlist: ran = wrap_var(ran) else: try: ran = UnsafeProxy(",".join(ran)) except TypeError: if isinstance(ran, list) and len(ran) == 1: ran = wrap_var(ran[0]) else: ran = wrap_var(ran) if self.cur_context: self.cur_context.unsafe = True return ran else: raise AnsibleError("lookup plugin (%s) not found" % name)
def _check_conditional(self, conditional, templar, all_vars): ''' This method does the low-level evaluation of each conditional set on this object, using jinja2 to wrap the conditionals for evaluation. ''' original = conditional if conditional is None or conditional == '': return True if conditional in all_vars and '-' not in text_type(all_vars[conditional]): conditional = all_vars[conditional] # make sure the templar is using the variables specified with this method templar.set_available_variables(variables=all_vars) try: conditional = templar.template(conditional) if not isinstance(conditional, text_type) or conditional == "": return conditional # a Jinja2 evaluation that results in something Python can eval! presented = "{%% if %s %%} True {%% else %%} False {%% endif %%}" % conditional conditional = templar.template(presented) val = conditional.strip() if val == "True": return True elif val == "False": return False else: raise AnsibleError("unable to evaluate conditional: %s" % original) except (AnsibleUndefinedVariable, UndefinedError) as e: # the templating failed, meaning most likely a # variable was undefined. If we happened to be # looking for an undefined variable, return True, # otherwise fail if "is undefined" in original: return True elif "is defined" in original: return False else: raise AnsibleError("error while evaluating conditional (%s): %s" % (original, e))
def run(self, terms, variables, **kwargs): anydict = False skip = False for term in terms: if isinstance(term, dict): anydict = True total_search = [] if anydict: for term in terms: if isinstance(term, dict): files = term.get('files', []) paths = term.get('paths', []) skip = boolean(term.get('skip', False)) filelist = files if isinstance(files, string_types): files = files.replace(',', ' ') files = files.replace(';', ' ') filelist = files.split(' ') pathlist = paths if paths: if isinstance(paths, string_types): paths = paths.replace(',', ' ') paths = paths.replace(':', ' ') paths = paths.replace(';', ' ') pathlist = paths.split(' ') if not pathlist: total_search = filelist else: for path in pathlist: for fn in filelist: f = os.path.join(path, fn) total_search.append(f) else: total_search.append(term) else: total_search = self._flatten(terms) for fn in total_search: try: fn = self._templar.template(fn) except (AnsibleUndefinedVariable, UndefinedError): continue # get subdir if set by task executor, default to files otherwise subdir = getattr(self, '_subdir', 'files') path = None path = self.find_file_in_search_path(variables, subdir, fn, ignore_missing=True) if path is not None: return [path] else: if skip: return [] else: raise AnsibleLookupError("No file was found when using with_first_found. Use the 'skip: true' option to allow this task to be skipped if no files are found")
def run(self, tmp=None, task_vars=None): if task_vars is None: task_vars = dict() for arg in self._task.args: if arg not in self.VALID_ARGS: return {"failed": True, "msg": "'%s' is not a valid option in debug" % arg} if 'msg' in self._task.args and 'var' in self._task.args: return {"failed": True, "msg": "'msg' and 'var' are incompatible options"} result = super(ActionModule, self).run(tmp, task_vars) verbosity = 0 # get task verbosity if 'verbosity' in self._task.args: verbosity = int(self._task.args['verbosity']) if verbosity <= self._display.verbosity: if 'msg' in self._task.args: result['msg'] = self._task.args['msg'] elif 'var' in self._task.args: try: results = self._templar.template(self._task.args['var'], convert_bare=True, fail_on_undefined=True, bare_deprecated=False) if results == self._task.args['var']: # if results is not str/unicode type, raise an exception if not isinstance(results, string_types): raise AnsibleUndefinedVariable # If var name is same as result, try to template it results = self._templar.template("{{" + results + "}}", convert_bare=True, fail_on_undefined=True) except AnsibleUndefinedVariable: results = "VARIABLE IS NOT DEFINED!" if isinstance(self._task.args['var'], (list, dict)): # If var is a list or dict, use the type as key to display result[to_text(type(self._task.args['var']))] = results else: result[self._task.args['var']] = results else: result['msg'] = 'Hello world!' # force flag to make debug output module always verbose result['_ansible_verbose_always'] = True else: result['skipped_reason'] = "Verbosity threshold not met." result['skipped'] = True return result
def run(self, terms, variables, **kwargs): anydict = False skip = False for term in terms: if isinstance(term, dict): anydict = True total_search = [] if anydict: for term in terms: if isinstance(term, dict): files = term.get('files', []) paths = term.get('paths', []) skip = boolean(term.get('skip', False), strict=False) filelist = files if isinstance(files, string_types): files = files.replace(',', ' ') files = files.replace(';', ' ') filelist = files.split(' ') pathlist = paths if paths: if isinstance(paths, string_types): paths = paths.replace(',', ' ') paths = paths.replace(':', ' ') paths = paths.replace(';', ' ') pathlist = paths.split(' ') if not pathlist: total_search = filelist else: for path in pathlist: for fn in filelist: f = os.path.join(path, fn) total_search.append(f) else: total_search.append(term) else: total_search = self._flatten(terms) for fn in total_search: try: fn = self._templar.template(fn) except (AnsibleUndefinedVariable, UndefinedError): continue # get subdir if set by task executor, default to files otherwise subdir = getattr(self, '_subdir', 'files') path = None path = self.find_file_in_search_path(variables, subdir, fn, ignore_missing=True) if path is not None: return [path] else: if skip: return [] else: raise AnsibleLookupError("No file was found when using with_first_found. Use the 'skip: true' option to allow this task to be skipped if no " "files are found")
