我们从Python开源项目中，提取了以下50个代码示例，用于说明如何使用ansible.errors.AnsibleFilterError()。
def get_netloc(url): """Return the netloc from a URL. If the input value is not a value URL the method will raise an Ansible filter exception. :param url: the URL to parse :type url: ``str`` :returns: ``str`` """ try: netloc = urlparse(url).netloc except Exception as exp: raise errors.AnsibleFilterError( 'Failed to return the netloc of: "%s"' % str(exp) ) else: return netloc
def get_netorigin(url): """Return the netloc from a URL. If the input value is not a value URL the method will raise an Ansible filter exception. :param url: the URL to parse :type url: ``str`` :returns: ``str`` """ try: parsed_url = urlparse(url) netloc = parsed_url.netloc scheme = parsed_url.scheme except Exception as exp: raise errors.AnsibleFilterError( 'Failed to return the netorigin of: "%s"' % str(exp) ) else: return '%s://%s' % (scheme, netloc)
def extract_by_prefix(values, prefix): """ This function takes a list as 'values' parameter and extract a first value from the list which contains prefix. The prefix is defined in parameter 'prefix' """ if values is None: raise errors.AnsibleFilterError('Values is not provided') if prefix is None: raise errors.AnsibleFilterError('Prefix is not provided') if not isinstance(values, list): raise errors.AnsibleFilterError('Wrong type for values') if not isinstance(prefix, str): raise errors.AnsibleFilterError('Wrong type for the prefix') if len(values) == 0: raise errors.AnsibleFilterError('Empty list. Nothing to extract') for v in values: if v.startswith(prefix): return v raise errors.AnsibleFilterError('Value not found')
def filter_by_prefix(values, prefix): """ This function takes a list as 'values' parameter and filters out all list values which contain prefix. The prefix is defined in parameter 'prefix' """ if values is None: raise errors.AnsibleFilterError('Values is not provided') if prefix is None: raise errors.AnsibleFilterError('Prefix is not provided') if not isinstance(values, list): raise errors.AnsibleFilterError('Wrong type for values') if not isinstance(prefix, str): raise errors.AnsibleFilterError('Wrong type for the prefix') return filter(lambda x: x.startswith(prefix), values)
def log(value, base): """ This function returns the logarithm of 'value' to the given base 'base' """ if value is None: raise errors.AnsibleFilterError('value is not provided') if base is None: raise errors.AnsibleFilterError('base is not provided') if not isinstance(value, int): raise errors.AnsibleFilterError('Wrong type for value') if not isinstance(base, int): raise errors.AnsibleFilterError('Wrong type for base') return math.log(value, base)
def _net_interface_type(context, name, inventory_hostname): """Return a string describing the network interface type. Possible types include 'ether', 'bridge', 'bond'. """ bridge_ports = net_bridge_ports(context, name, inventory_hostname) bond_slaves = net_bond_slaves(context, name, inventory_hostname) if bridge_ports is not None and bond_slaves is not None: raise errors.AnsibleFilterError( "Network %s on host %s has both bridge ports and bond slaves " "defined" % (name, _get_hostvar(context, 'inventory_hostname', inventory_hostname))) if bridge_ports is None and bond_slaves is None: return 'ether' if bridge_ports is not None: return 'bridge' if bond_slaves is not None: return 'bond'
def previous_nth_usable(value, offset): try: vtype = ipaddr(value, 'type') if vtype == 'address': v = ipaddr(value, 'cidr') elif vtype == 'network': v = ipaddr(value, 'subnet') v = netaddr.IPNetwork(v) except: return False if type(offset) != int: raise errors.AnsibleFilterError('Must pass in an interger') if v.size > 1: first_usable, last_usable = _first_last(v) nth_ip = int(netaddr.IPAddress(int(v.ip) - offset)) if nth_ip >= first_usable and nth_ip <= last_usable: return str(netaddr.IPAddress(int(v.ip) - offset))
def rand(environment, end, start=None, step=None, seed=None): if seed is None: r = SystemRandom() else: r = Random(seed) if isinstance(end, integer_types): if not start: start = 0 if not step: step = 1 return r.randrange(start, end, step) elif hasattr(end, '__iter__'): if start or step: raise errors.AnsibleFilterError('start and step can only be used with integer values') return r.choice(end) else: raise errors.AnsibleFilterError('random can only be used on sequences and integers')
def changed(result): ''' Test if task result yields changed ''' if not isinstance(result, MutableMapping): raise errors.AnsibleFilterError("|changed expects a dictionary") if 'changed' not in result: changed = False if ( 'results' in result and # some modules return a 'results' key isinstance(result['results'], MutableSequence) and isinstance(result['results'][0], MutableMapping) ): for res in result['results']: if res.get('changed', False): changed = True break else: changed = result.get('changed', False) return changed
def get_attr(data, attribute=None): """ This looks up dictionary attributes of the form a.b.c and returns the value. If the key isn't present, None is returned. Ex: data = {'a': {'b': {'c': 5}}} attribute = "a.b.c" returns 5 """ if not attribute: raise errors.AnsibleFilterError("|failed expects attribute to be set") ptr = data for attr in attribute.split('.'): if attr in ptr: ptr = ptr[attr] else: ptr = None break return ptr
def oo_select_keys_from_list(data, keys): """ This returns a list, which contains the value portions for the keys Ex: data = { 'a':1, 'b':2, 'c':3 } keys = ['a', 'c'] returns [1, 3] """ if not isinstance(data, list): raise errors.AnsibleFilterError("|failed expects to filter on a list") if not isinstance(keys, list): raise errors.AnsibleFilterError("|failed expects first param is a list") # Gather up the values for the list of keys passed in retval = [FilterModule.oo_select_keys(item, keys) for item in data] return FilterModule.oo_flatten(retval)
def oo_select_keys(data, keys): """ This returns a list, which contains the value portions for the keys Ex: data = { 'a':1, 'b':2, 'c':3 } keys = ['a', 'c'] returns [1, 3] """ if not isinstance(data, Mapping): raise errors.AnsibleFilterError("|failed expects to filter on a dict or object") if not isinstance(keys, list): raise errors.AnsibleFilterError("|failed expects first param is a list") # Gather up the values for the list of keys passed in retval = [data[key] for key in keys if key in data] return retval
def oo_filter_list(data, filter_attr=None): """ This returns a list, which contains all items where filter_attr evaluates to true Ex: data = [ { a: 1, b: True }, { a: 3, b: False }, { a: 5, b: True } ] filter_attr = 'b' returns [ { a: 1, b: True }, { a: 5, b: True } ] """ if not isinstance(data, list): raise errors.AnsibleFilterError("|failed expects to filter on a list") if not isinstance(filter_attr, basestring): raise errors.AnsibleFilterError("|failed expects filter_attr is a str or unicode") # Gather up the values for the list of keys passed in return [x for x in data if filter_attr in x and x[filter_attr]]
def oo_31_rpm_rename_conversion(rpms, openshift_version=None): """ Filters a list of 3.0 rpms and return the corresponding 3.1 rpms names with proper version (if provided) If 3.1 rpms are passed in they will only be augmented with the correct version. This is important for hosts that are running both Masters and Nodes. """ if not isinstance(rpms, list): raise errors.AnsibleFilterError("failed expects to filter on a list") if openshift_version is not None and not isinstance(openshift_version, basestring): raise errors.AnsibleFilterError("failed expects openshift_version to be a string") rpms_31 = [] for rpm in rpms: if not 'atomic' in rpm: rpm = rpm.replace("openshift", "atomic-openshift") if openshift_version: rpm = rpm + openshift_version rpms_31.append(rpm) return rpms_31
def oo_pods_match_component(pods, deployment_type, component): """ Filters a list of Pods and returns the ones matching the deployment_type and component """ if not isinstance(pods, list): raise errors.AnsibleFilterError("failed expects to filter on a list") if not isinstance(deployment_type, basestring): raise errors.AnsibleFilterError("failed expects deployment_type to be a string") if not isinstance(component, basestring): raise errors.AnsibleFilterError("failed expects component to be a string") image_prefix = 'openshift/origin-' if deployment_type in ['enterprise', 'online', 'openshift-enterprise']: image_prefix = 'openshift3/ose-' elif deployment_type == 'atomic-enterprise': image_prefix = 'aep3_beta/aep-' matching_pods = [] image_regex = image_prefix + component + r'.*' for pod in pods: for container in pod['spec']['containers']: if re.search(image_regex, container['image']): matching_pods.append(pod) break # stop here, don't add a pod more than once return matching_pods
def oo_image_tag_to_rpm_version(version, include_dash=False): """ Convert an image tag string to an RPM version if necessary Empty strings and strings that are already in rpm version format are ignored. Also remove non semantic version components. Ex. v3.2.0.10 -> -3.2.0.10 v1.2.0-rc1 -> -1.2.0 """ if not isinstance(version, basestring): raise errors.AnsibleFilterError("|failed expects a string or unicode") if version.startswith("v"): version = version[1:] # Strip release from requested version, we no longer support this. version = version.split('-')[0] if include_dash and version and not version.startswith("-"): version = "-" + version return version
def validate_idp_list(idp_list, openshift_version, deployment_type): ''' validates a list of idps ''' login_providers = [x.name for x in idp_list if x.login] multiple_logins_unsupported = False if len(login_providers) > 1: if deployment_type in ['enterprise', 'online', 'atomic-enterprise', 'openshift-enterprise']: if LooseVersion(openshift_version) < LooseVersion('3.2'): multiple_logins_unsupported = True if deployment_type in ['origin']: if LooseVersion(openshift_version) < LooseVersion('1.2'): multiple_logins_unsupported = True if multiple_logins_unsupported: raise errors.AnsibleFilterError("|failed multiple providers are " "not allowed for login. login " "providers: {0}".format(', '.join(login_providers))) names = [x.name for x in idp_list] if len(set(names)) != len(names): raise errors.AnsibleFilterError("|failed more than one provider configured with the same name") for idp in idp_list: idp.validate()
def validate(self): ''' validate this idp instance ''' IdentityProviderBase.validate(self) if not isinstance(self.provider['attributes'], dict): raise errors.AnsibleFilterError("|failed attributes for provider " "{0} must be a dictionary".format(self.__class__.__name__)) attrs = ['id', 'email', 'name', 'preferredUsername'] for attr in attrs: if attr in self.provider['attributes'] and not isinstance(self.provider['attributes'][attr], list): raise errors.AnsibleFilterError("|failed {0} attribute for " "provider {1} must be a list".format(attr, self.__class__.__name__)) unknown_attrs = set(self.provider['attributes'].keys()) - set(attrs) if len(unknown_attrs) > 0: raise errors.AnsibleFilterError("|failed provider {0} has unknown " "attributes: {1}".format(self.__class__.__name__, ', '.join(unknown_attrs)))
def translate_idps(idps, api_version, openshift_version, deployment_type): ''' Translates a list of dictionaries into a valid identityProviders config ''' idp_list = [] if not isinstance(idps, list): raise errors.AnsibleFilterError("|failed expects to filter on a list of identity providers") for idp in idps: if not isinstance(idp, dict): raise errors.AnsibleFilterError("|failed identity providers must be a list of dictionaries") cur_module = sys.modules[__name__] idp_class = getattr(cur_module, idp['kind'], None) idp_inst = idp_class(api_version, idp) if idp_class is not None else IdentityProviderBase(api_version, idp) idp_inst.set_provider_items() idp_list.append(idp_inst) IdentityProviderBase.validate_idp_list(idp_list, openshift_version, deployment_type) return yaml.safe_dump([idp.to_dict() for idp in idp_list], default_flow_style=False)
def validate_pcs_cluster(data, masters=None): ''' Validates output from "pcs status", ensuring that each master provided is online. Ex: data = ('...', 'PCSD Status:', 'master1.example.com: Online', 'master2.example.com: Online', 'master3.example.com: Online', '...') masters = ['master1.example.com', 'master2.example.com', 'master3.example.com'] returns True ''' if not issubclass(type(data), basestring): raise errors.AnsibleFilterError("|failed expects data is a string or unicode") if not issubclass(type(masters), list): raise errors.AnsibleFilterError("|failed expects masters is a list") valid = True for master in masters: if "{0}: Online".format(master) not in data: valid = False return valid
def oo_htpasswd_users_from_file(file_contents): ''' return a dictionary of htpasswd users from htpasswd file contents ''' htpasswd_entries = {} if not isinstance(file_contents, basestring): raise errors.AnsibleFilterError("failed, expects to filter on a string") for line in file_contents.splitlines(): user = None passwd = None if len(line) == 0: continue if ':' in line: user, passwd = line.split(':', 1) if user is None or len(user) == 0 or passwd is None or len(passwd) == 0: error_msg = "failed, expects each line to be a colon separated string representing the user and passwd" raise errors.AnsibleFilterError(error_msg) htpasswd_entries[user] = passwd return htpasswd_entries
def oo_ami_selector(data, image_name): """ This takes a list of amis and an image name and attempts to return the latest ami. """ if not isinstance(data, list): raise errors.AnsibleFilterError("|failed expects first param is a list") if not data: return None else: if image_name is None or not image_name.endswith('_*'): ami = sorted(data, key=itemgetter('name'), reverse=True)[0] return ami['ami_id'] else: ami_info = [(ami, ami['name'].split('_')[-1]) for ami in data] ami = sorted(ami_info, key=itemgetter(1), reverse=True)[0][0] return ami['ami_id']
def oo_openshift_env(hostvars): ''' Return facts which begin with "openshift_" and translate legacy facts to their openshift_env counterparts. Ex: hostvars = {'openshift_fact': 42, 'theyre_taking_the_hobbits_to': 'isengard'} returns = {'openshift_fact': 42} ''' if not issubclass(type(hostvars), dict): raise errors.AnsibleFilterError("|failed expects hostvars is a dict") facts = {} regex = re.compile('^openshift_.*') for key in hostvars: if regex.match(key): facts[key] = hostvars[key] migrations = {'openshift_router_selector': 'openshift_hosted_router_selector', 'openshift_registry_selector': 'openshift_hosted_registry_selector'} for old_fact, new_fact in migrations.iteritems(): if old_fact in facts and new_fact not in facts: facts[new_fact] = facts[old_fact] return facts
