a-c-m
/
ansible.utils
mirror of https://github.com/ansible-collections/ansible.utils.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

init code

pull/156/head
KB-perByte 2022-04-04 17:55:01 +05:30
parent 5eba64f8de
commit 66bd592df4
No known key found for this signature in database
GPG Key ID: F49472BE1744C3D5
2 changed files with 475 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

349
plugins/filter/consolidate.py Normal file
View File

@ -0,0 +1,349 @@
#
# -*- coding: utf-8 -*-
# Copyright 2021 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
#
"""
The consolidate filter plugin
"""
from __future__ import absolute_import, division, print_function
__metaclass__ = type
DOCUMENTATION = """
name: consolidate
author: Sagar Paul (@KB-perByte)
version_added: "2.5.0"
short_description: Keep specific keys from a data recursively.
description:
- This plugin keep only specified keys from a provided data recursively.
- Matching parameter defaults to equals unless C(matching_parameter) is explicitly mentioned.
- Using the parameters below- C(data|ansible.utils.keep_keys(target([....])))
options:
data_source:
description:
- This option represents a list of dictionaries or a dictionary with any level of nesting data.
- For example C(config_data|ansible.utils.keep_keys(target([....]))), in this case C(config_data) represents this option.
type: list
elements: dict
suboptions:
data:
description: Specify the target keys to keep in list format.
type: raw
match_key:
description: Specify the target keys to keep in list format.
type: str
prefix:
description: Specify the target keys to keep in list format.
type: str
fail_missing_match_key:
description: Specify the target keys to keep in list format.
type: bool
fail_missing_match_value:
description: Specify the target keys to keep in list format.
type: bool
fail_duplicate:
description: Specify the matching configuration of target keys and data attributes.
type: bool
"""
EXAMPLES = r"""
##example.yaml
interfaces:
- name: eth0
enabled: true
duplex: auto
speed: auto
note:
- Connected green wire
- name: eth1
description: Configured by Ansible - Interface 1
mtu: 1500
speed: auto
duplex: auto
enabled: true
note:
- Connected blue wire
- Configured by Paul
vifs:
- vlan_id: 100
description: Eth1 - VIF 100
mtu: 400
enabled: true
comment: Needs reconfiguration
- vlan_id: 101
description: Eth1 - VIF 101
enabled: true
- name: eth2
description: Configured by Ansible - Interface 2 (ADMIN DOWN)
mtu: 600
enabled: false
##Playbook
vars_files:
- "example.yaml"
tasks:
- name: keep selective keys from dict/list of dict data
ansible.builtin.set_fact:
data: "{{ interfaces }}"
- debug:
msg: "{{ data|ansible.utils.keep_keys(target=['description', 'name', 'mtu', 'duplex', 'enabled', 'vifs', 'vlan_id']) }}"
##Output
# TASK [keep selective keys from python dict/list of dict] ****************************************************************************************
# ok: [localhost] => {
# "ansible_facts": {
# "data": [
# {
# "duplex": "auto",
# "enabled": true,
# "name": "eth0",
# "note": [
# "Connected green wire"
# ],
# "speed": "auto"
# },
# {
# "description": "Configured by Ansible - Interface 1",
# "duplex": "auto",
# "enabled": true,
# "mtu": 1500,
# "name": "eth1",
# "note": [
# "Connected blue wire",
# "Configured by Paul"
# ],
# "speed": "auto",
# "vifs": [
# {
# "comment": "Needs reconfiguration",
# "description": "Eth1 - VIF 100",
# "enabled": true,
# "mtu": 400,
# "vlan_id": 100
# },
# {
# "description": "Eth1 - VIF 101",
# "enabled": true,
# "vlan_id": 101
# }
# ]
# },
# {
# "description": "Configured by Ansible - Interface 2 (ADMIN DOWN)",
# "enabled": false,
# "mtu": 600,
# "name": "eth2"
# }
# ]
# },
# "changed": false
# }
# Read vars_file 'example.yaml'
# TASK [debug] *************************************************************************************************************
# ok: [localhost] => {
# "msg": [
# {
# "duplex": "auto",
# "enabled": true,
# "name": "eth0"
# },
# {
# "description": "Configured by Ansible - Interface 1",
# "duplex": "auto",
# "enabled": true,
# "mtu": 1500,
# "name": "eth1",
# "vifs": [
# {
# "description": "Eth1 - VIF 100",
# "enabled": true,
# "mtu": 400,
# "vlan_id": 100
# },
# {
# "description": "Eth1 - VIF 101",
# "enabled": true,
# "vlan_id": 101
# }
# ]
# },
# {
# "description": "Configured by Ansible - Interface 2 (ADMIN DOWN)",
# "enabled": false,
# "mtu": 600,
# "name": "eth2"
# }
# ]
# }
##example.yaml
interfaces:
- name: eth0
enabled: true
duplex: auto
speed: auto
note:
- Connected green wire
- name: eth1
description: Configured by Ansible - Interface 1
mtu: 1500
speed: auto
duplex: auto
enabled: true
note:
- Connected blue wire
- Configured by Paul
vifs:
- vlan_id: 100
description: Eth1 - VIF 100
mtu: 400
enabled: true
comment: Needs reconfiguration
- vlan_id: 101
description: Eth1 - VIF 101
enabled: true
- name: eth2
description: Configured by Ansible - Interface 2 (ADMIN DOWN)
mtu: 600
enabled: false
##Playbook
vars_files:
- "example.yaml"
tasks:
- name: keep selective keys from dict/list of dict data
ansible.builtin.set_fact:
data: "{{ interfaces }}"
- debug:
msg: "{{ data|ansible.utils.keep_keys(target=['desc', 'name'], matching_parameter= 'starts_with') }}"
##Output
# TASK [keep selective keys from python dict/list of dict] **************************
# ok: [localhost] => {
# "ansible_facts": {
# "data": [
# {
# "duplex": "auto",
# "enabled": true,
# "name": "eth0",
# "note": [
# "Connected green wire"
# ],
# "speed": "auto"
# },
# {
# "description": "Configured by Ansible - Interface 1",
# "duplex": "auto",
# "enabled": true,
# "mtu": 1500,
# "name": "eth1",
# "note": [
# "Connected blue wire",
# "Configured by Paul"
# ],
# "speed": "auto",
# "vifs": [
# {
# "comment": "Needs reconfiguration",
# "description": "Eth1 - VIF 100",
# "enabled": true,
# "mtu": 400,
# "vlan_id": 100
# },
# {
# "description": "Eth1 - VIF 101",
# "enabled": true,
# "vlan_id": 101
# }
# ]
# },
# {
# "description": "Configured by Ansible - Interface 2 (ADMIN DOWN)",
# "enabled": false,
# "mtu": 600,
# "name": "eth2"
# }
# ]
# },
# "changed": false
# }
# Read vars_file 'example.yaml'
# TASK [debug] **********************************************************************************
# ok: [localhost] => {
# "msg": [
# {
# "name": "eth0"
# },
# {
# "description": "Configured by Ansible - Interface 1",
# "name": "eth1",
# "vifs": [
# {
# "description": "Eth1 - VIF 100"
# },
# {
# "description": "Eth1 - VIF 101"
# }
# ]
# },
# {
# "description": "Configured by Ansible - Interface 2 (ADMIN DOWN)",
# "name": "eth2"
# }
# ]
# }
"""
from ansible.errors import AnsibleFilterError
from ansible_collections.ansible.utils.plugins.plugin_utils.consolidate import (
consolidate,
)
from ansible_collections.ansible.utils.plugins.module_utils.common.argspec_validate import (
AnsibleArgSpecValidator,
)
try:
from jinja2.filters import pass_environment
except ImportError:
from jinja2.filters import environmentfilter as pass_environment
import debugpy
debugpy.listen(3000)
debugpy.wait_for_client()
@pass_environment
def _consolidate(*args, **kwargs):
"""keep specific keys from a data recursively"""
keys = [
"data_source",
"fail_missing_match_key",
"fail_missing_match_value",
"fail_duplicate",
]
data = dict(zip(keys, args[1:]))
data.update(kwargs)
aav = AnsibleArgSpecValidator(data=data, schema=DOCUMENTATION, name="consolidate")
valid, errors, updated_data = aav.validate()
if not valid:
raise AnsibleFilterError(errors)
return consolidate(**updated_data)
class FilterModule(object):
"""keep_keys"""
def filters(self):
"""a mapping of filter names to functions"""
return {"consolidate": _consolidate}

126
plugins/plugin_utils/consolidate.py Normal file
View File

@ -0,0 +1,126 @@
#
# -*- coding: utf-8 -*-
# Copyright 2021 Red Hat
# GNU General Public License v3.0+
# (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)
#
"""
The keep_keys plugin code
"""
from __future__ import absolute_import, division, print_function
__metaclass__ = type
from ansible.errors import AnsibleFilterError
import itertools
def _raise_error(filter, msg):
"""Raise an error message, prepend with filter name
:param msg: The message
:type msg: str
:raises: AnsibleError
"""
error = f"Error when using plugin 'consolidate': '{filter}' reported {msg}"
raise AnsibleFilterError(error)
def fail_on_filter(validator_func):
def update_err(*args, **kwargs):
res, err = validator_func(*args, **kwargs)
if err.get("match_key_err"):
_raise_error("fail_missing_match_key", ", ".join(err["match_key_err"]))
if err.get("match_val_err"):
_raise_error("fail_missing_match_value", ", ".join(err["match_val_err"]))
if err.get("duplicate_err"):
_raise_error("fail_duplicate", ", ".join(err["duplicate_err"]))
return res
return update_err
@fail_on_filter
def check_missing_match_key_duplicate(
data_sources, fail_missing_match_key, fail_duplicate
):
"""Validate the operation
:param operation: The operation
:type operation: str
:raises: AnsibleFilterError
"""
results, errors_match_key, errors_duplicate = [], [], []
# Check for missing and duplicate match key
for ds_idx, data_source in enumerate(data_sources):
match_key = data_source["match_key"]
ds_values = []
for dd_idx, data_dict in enumerate(data_source["data"]):
try:
ds_values.append(data_dict[match_key])
except KeyError:
if fail_missing_match_key:
errors_match_key.append(
f"Missing match key '{match_key}' in data source {ds_idx} in list entry {dd_idx}"
)
continue
if sorted(set(ds_values)) != sorted(ds_values) and fail_duplicate:
errors_duplicate.append(f"Duplicate values in data source {ds_idx}")
results.append(set(ds_values))
return results, {
"match_key_err": errors_match_key,
"duplicate_err": errors_duplicate,
}
@fail_on_filter
def check_missing_match_values(results, fail_missing_match_value):
errors_match_values = []
all_values = set(itertools.chain.from_iterable(results))
if fail_missing_match_value:
for ds_idx, ds_values in enumerate(results):
missing_match = all_values - ds_values
if missing_match:
errors_match_values.append(
f"Missing match value {', '.join(missing_match)} in data source {ds_idx}"
)
return all_values, {"match_val_err": errors_match_values}
def consolidate_facts(data_sources, all_values):
consolidated_facts = {}
for data_source in data_sources:
match_key = data_source["match_key"]
source = data_source["prefix"]
data_dict = {d[match_key]: d for d in data_source["data"] if match_key in d}
for value in sorted(all_values):
if value not in consolidated_facts:
consolidated_facts[value] = {}
consolidated_facts[value][source] = data_dict.get(value, {})
return consolidated_facts
def consolidate(
data_source,
fail_missing_match_key=False,
fail_missing_match_value=False,
fail_duplicate=False,
):
"""keep selective keys recursively from a given data"
:param data: The data passed in (data|keep_keys(...))
:type data: raw
:param target: List of keys on with operation is to be performed
:type data: list
:type elements: string
:param matching_parameter: matching type of the target keys with data keys
:type data: str
"""
# write code here
key_sets = check_missing_match_key_duplicate(
data_source, fail_missing_match_key, fail_duplicate
)
key_vals = check_missing_match_values(key_sets, fail_missing_match_value)
datapr = consolidate_facts(data_source, key_vals)
return datapr