mirror of
https://github.com/ansible-collections/kubernetes.core.git
synced 2026-03-30 23:33:03 +00:00
08a3d951d0
Move module dependency functions outside of module SUMMARY This moves the has_at_least and requires functions that had been on the module to top level functions. The functions on the module now call these with a few added bits of functionality. Moving these functions to the top level and removing their requirement on having a module makes them usable in situations where we may not yet have a module, such as during client creation. ISSUE TYPE COMPONENT NAME ADDITIONAL INFORMATION Reviewed-by: Alina Buzachis <None> Reviewed-by: None <None>
153 lines
4.2 KiB
Python
153 lines
4.2 KiB
Python
from distutils.version import LooseVersion
|
|
from typing import Optional
|
|
|
|
from ansible.module_utils.basic import AnsibleModule
|
|
from ansible.module_utils.basic import missing_required_lib
|
|
from ansible.module_utils.common.text.converters import to_text
|
|
|
|
|
|
class AnsibleK8SModule:
|
|
"""A base module class for K8S modules.
|
|
|
|
This class should be used instead of directly using AnsibleModule. If there
|
|
is a need for other methods or attributes to be proxied, they can be added
|
|
here.
|
|
"""
|
|
|
|
default_settings = {
|
|
"check_k8s": True,
|
|
"check_pyyaml": True,
|
|
"module_class": AnsibleModule,
|
|
}
|
|
|
|
def __init__(self, **kwargs) -> None:
|
|
local_settings = {}
|
|
for key in AnsibleK8SModule.default_settings:
|
|
try:
|
|
local_settings[key] = kwargs.pop(key)
|
|
except KeyError:
|
|
local_settings[key] = AnsibleK8SModule.default_settings[key]
|
|
self.settings = local_settings
|
|
|
|
self._module = self.settings["module_class"](**kwargs)
|
|
|
|
if self.settings["check_k8s"]:
|
|
self.requires("kubernetes")
|
|
self.has_at_least("kubernetes", "12.0.0", warn=True)
|
|
|
|
if self.settings["check_pyyaml"]:
|
|
self.requires("pyyaml")
|
|
|
|
@property
|
|
def check_mode(self):
|
|
return self._module.check_mode
|
|
|
|
@property
|
|
def _diff(self):
|
|
return self._module._diff
|
|
|
|
@property
|
|
def _name(self):
|
|
return self._module._name
|
|
|
|
@property
|
|
def params(self):
|
|
return self._module.params
|
|
|
|
def warn(self, *args, **kwargs):
|
|
return self._module.warn(*args, **kwargs)
|
|
|
|
def deprecate(self, *args, **kwargs):
|
|
return self._module.deprecate(*args, **kwargs)
|
|
|
|
def debug(self, *args, **kwargs):
|
|
return self._module.debug(*args, **kwargs)
|
|
|
|
def exit_json(self, *args, **kwargs):
|
|
return self._module.exit_json(*args, **kwargs)
|
|
|
|
def fail_json(self, *args, **kwargs):
|
|
return self._module.fail_json(*args, **kwargs)
|
|
|
|
def has_at_least(
|
|
self, dependency: str, minimum: Optional[str] = None, warn: bool = False
|
|
) -> bool:
|
|
supported = has_at_least(dependency, minimum)
|
|
if not supported and warn:
|
|
self.warn(
|
|
"{0}<{1} is not supported or tested. Some features may not work.".format(
|
|
dependency, minimum
|
|
)
|
|
)
|
|
return supported
|
|
|
|
def requires(
|
|
self,
|
|
dependency: str,
|
|
minimum: Optional[str] = None,
|
|
reason: Optional[str] = None,
|
|
) -> None:
|
|
try:
|
|
requires(dependency, minimum, reason=reason)
|
|
except Exception as e:
|
|
self.fail_json(msg=to_text(e))
|
|
|
|
|
|
def gather_versions() -> dict:
|
|
versions = {}
|
|
try:
|
|
import jsonpatch
|
|
|
|
versions["jsonpatch"] = jsonpatch.__version__
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
import kubernetes
|
|
|
|
versions["kubernetes"] = kubernetes.__version__
|
|
except ImportError:
|
|
pass
|
|
|
|
try:
|
|
import yaml
|
|
|
|
versions["pyyaml"] = yaml.__version__
|
|
except ImportError:
|
|
pass
|
|
|
|
return versions
|
|
|
|
|
|
def has_at_least(dependency: str, minimum: Optional[str] = None) -> bool:
|
|
"""Check if a specific dependency is present at a minimum version.
|
|
|
|
If a minimum version is not specified it will check only that the
|
|
dependency is present.
|
|
"""
|
|
dependencies = gather_versions()
|
|
current = dependencies.get(dependency)
|
|
if current is not None:
|
|
if minimum is None:
|
|
return True
|
|
supported = LooseVersion(current) >= LooseVersion(minimum)
|
|
return supported
|
|
return False
|
|
|
|
|
|
def requires(
|
|
dependency: str, minimum: Optional[str] = None, reason: Optional[str] = None
|
|
) -> None:
|
|
"""Fail if a specific dependency is not present at a minimum version.
|
|
|
|
If a minimum version is not specified it will require only that the
|
|
dependency is present. This function raises an exception when the
|
|
dependency is not found at the required version.
|
|
"""
|
|
if not has_at_least(dependency, minimum):
|
|
if minimum is not None:
|
|
lib = "{0}>={1}".format(dependency, minimum)
|
|
else:
|
|
lib = dependency
|
|
raise Exception(missing_required_lib(lib, reason=reason))
|