文件操作 - utils.py

文件: /usr/lib/python3.12/site-packages/ansible_collections/sensu/sensu_go/plugins/module_utils/utils.py

编辑文件内容

# -*- coding: utf-8 -*-
# Copyright: (c) 2019, XLAB Steampunk <steampunk@xlab.si>
#
# GNU General Public License v3.0+ (see COPYING or https://www.gnu.org/licenses/gpl-3.0.txt)

from __future__ import absolute_import, division, print_function
__metaclass__ = type

from ansible.module_utils.six.moves.urllib.parse import quote

from . import errors

def do_differ(current, desired, *ignored_keys):
    if "metadata" in desired:
        # We treat metadata as any regular dict but ignore the created_by
        # value since this piece of data is autogenerated on the backend and
        # must be skipped in the comparison.
        if do_differ(current["metadata"], desired["metadata"], "created_by"):
            return True

ignored_keys = ignored_keys + ("metadata",)

for key, value in desired.items():
        if key in ignored_keys:
            continue

if value != current.get(key):
            return True

return False

def do_differ_v1(current, desired, *ignored_keys):
    # Default comparator for v1 API (mostly enterprise features)

if "metadata" in desired:
        # We treat metadata as any regular dict but ignore the created_by
        # value since this piece of data is autogenerated on the backend and
        # must be skipped in the comparison.
        if do_differ(current["metadata"], desired["metadata"], "created_by"):
            return True

for key, value in desired.get("spec", {}).items():
        if key in ignored_keys:
            continue

if value != current["spec"].get(key):
            return True

return False

def sync(state, client, path, payload, check_mode, compare=do_differ):
    remote_object = get(client, path)

if state == "absent" and remote_object is None:
        return False, None

if state == "absent":
        if not check_mode:
            delete(client, path)
        return True, None

# Making sure remote_object is present from here on

if remote_object is None or compare(remote_object, payload):
        if check_mode:
            return True, payload
        put(client, path, payload)
        return True, get(client, path)

return False, remote_object

def sync_v1(state, client, path, payload, check_mode, compare=do_differ_v1):
    changed, result = sync(state, client, path, payload, check_mode, compare)
    return changed, convert_v1_to_v2_response(result)

def _abort(msg, *args, **kwargs):
    raise errors.SyncError(msg.format(*args, **kwargs))

def get(client, path):
    resp = client.get(path)
    if resp.status not in (200, 404):
        _abort(
            "GET {0} failed with status {1}: {2}", path, resp.status, resp.data,
        )
    if resp.status == 200 and resp.json is None:
        _abort("Server returned invalid JSON {0}", resp.data)
    return resp.json

def delete(client, path):
    resp = client.delete(path)
    if resp.status != 204:
        _abort(
            "DELETE {0} failed with status {1}: {2}",
            path, resp.status, resp.data,
        )
    return None

def put(client, path, payload):
    resp = client.put(path, payload)
    if resp.status not in (200, 201):
        _abort(
            "PUT {0} failed with status {1}: {2}",
            path, resp.status, resp.data,
        )
    return None

def dict_to_single_item_dicts(data):
    return [{k: v} for k, v in data.items()]

def single_item_dicts_to_dict(data):
    result = {}
    for item in data:
        (k, v), = item.items()
        result[k] = v
    return result

def dict_to_key_value_strings(data):
    return ["{0}={1}".format(k, v) for k, v in data.items()]

def build_url_path(api_group, api_version, namespace, *parts):
    prefix = "/api/{0}/{1}/".format(api_group, api_version)
    if namespace:
        prefix += "namespaces/{0}/".format(quote(namespace, safe=""))
    return prefix + "/".join(quote(p, safe="") for p in parts if p)

def build_core_v2_path(namespace, *parts):
    return build_url_path("core", "v2", namespace, *parts)

def prepare_result_list(result):
    if isinstance(result, list):
        return result
    return [] if result is None else [result]

def convert_v1_to_v2_response(response):
    # dict(metadata=<meta>, spec=dict(a=1)) -> dict(metadata=<meta>, a=1)

if not response:
        return response

if "metadata" not in response:
        return response["spec"]

# Move metadata key into the spec.
    return dict(response["spec"], metadata=response["metadata"])

def do_secrets_differ(current, desired):
    return set(
        (c["name"], c["secret"]) for c in (current.get("secrets") or [])
    ) != set(
        (d["name"], d["secret"]) for d in (desired.get("secrets") or [])
    )

def deprecate(module, msg, version):
    try:
        module.deprecate(msg, version=version, collection_name="sensu.sensu_go")
    except TypeError:
        # Ansible < 2.9.10 does not support collection_name kwarg. Output at
        # least msg and version of collection when things will stop working.
        module.deprecate(msg, version=version)

文件操作 - utils.py

文件: /usr/lib/python3.12/site-packages/ansible_collections/sensu/sensu_go/plugins/module_utils/utils.py

编辑文件内容

修改文件时间

删除文件