ansible-module/plugins/module_utils/generic.py

176 Zeilen
5.5 KiB
Python

import pathlib
import warnings
from typing import Any, Callable, Dict, Optional, Sequence, Tuple, Type, Union
__all__ = (
"Types",
"SYSTEMD_SERVICE_CONFIG",
"SYSTEMD_NETWORK_CONFIG",
"SYSTEMD_CONFIG_ROOT",
"systemdbool",
"AnsibleParameter",
)
SYSTEMD_CONFIG_ROOT = pathlib.Path("/etc/systemd")
SYSTEMD_NETWORK_CONFIG = SYSTEMD_CONFIG_ROOT / "network"
SYSTEMD_SERVICE_CONFIG = SYSTEMD_CONFIG_ROOT / "system"
AnsibleParameter = Dict[str, Any]
def wrap_func(func, **updates):
def wrapper(*args, **kwargs):
return func(*args, **kwargs)
attrs = frozenset(
('__module__', '__name__', '__qualname__', '__doc__', '__annotations__', '__type_params__')
) - frozenset(updates.keys())
for attr in attrs:
try:
value = getattr(func, attr)
except AttributeError:
pass
else:
setattr(wrapper, attr, value)
for attr, value in updates.items():
setattr(wrapper, attr, value)
wrapper.__dict__.update(func.__dict__)
setattr(wrapper, "__wrapped__", func)
return wrapper
GENERIC_DOC = """Returns an dictionary for the Ansible {type} type."""
def default(name: str):
def wrapped(
required: bool = False,
help: Optional[str] = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
) -> AnsibleParameter:
option: AnsibleParameter = dict(type=name, required=required)
if choices is not None:
option["choices"] = choices
if default is not None:
option["default"] = default
if help is not None:
option["description"] = help.split("\n")
return option
return wrapped
class meta(type):
def __new__(cls, clsname, bases, attrs):
types = frozenset(
(
"str",
"bool",
"int",
"float",
"path",
"raw",
"jsonarg",
"json",
"bytes",
"dict",
"list",
"bits",
)
)
for attr in types - set(attrs.keys()):
attrs[attr] = wrap_func(
default(attr), __doc__=GENERIC_DOC.format(type=attr), __name__=attr, __qualname__=f"{clsname}.{attr}"
)
attrs["__slots__"] = ()
return super().__new__(cls, clsname, bases, attrs)
class Types(metaclass=meta):
def list( # type: ignore[misc]
elements: Union[Type[object], str, AnsibleParameter],
required: bool = False,
help: Optional[str] = None,
) -> AnsibleParameter:
"""Wrapper for the Ansible list type
Args:
elements: The type of the elements
required: if the item is absolutly required
help: an helptext for the ansible-doc
"""
option: AnsibleParameter = dict(type="list", required=required)
if not isinstance(elements, (str, dict)):
option["elements"] = elements.__name__
elif isinstance(elements, dict):
option["elements"] = elements["type"]
if elements["type"] == "dict":
option["options"] = dict()
for name, value in elements["option"].items():
option["options"][name] = value
if "description" not in option["options"][name]:
warnings.warn( # pragma: nocover
f"helptext of option {name} is unset."
" Ansible requires suboptions to have an documentation"
)
if help is not None:
option["description"] = help.split("\n")
return option
def dict(required: bool = False, help: Optional[str] = None, **options: AnsibleParameter) -> AnsibleParameter: # type: ignore[misc]
"""Wrapper for the Ansible dict type
Args:
required: if the item is absolutly required
help: an helptext for the ansible-doc
options: The individual options that this parameter has
"""
option: AnsibleParameter = dict(type="dict", required=required)
option["options"] = options
if help is not None:
option["description"] = help.split("\n")
return option
def systemdbool(b: Union[bool, str]) -> str:
"""Converts values into things systemd can parse"""
if b is True:
return "yes"
elif b is False:
return "no"
return b
def modspec(
argument_spec: Dict[str, Dict[str, Any]],
mutually_exclusive: Sequence[Tuple[str, ...]] = (),
required_together: Sequence[Tuple[str, ...]] = (),
required_one_of: Sequence[Tuple[str, ...]] = (),
required_if: Sequence[Union[Tuple[str, Any, Tuple[str, ...]], Tuple[str, Any, Tuple[str, ...], bool]]] = (),
required_by: Dict[str, Union[str, Tuple[str, ...]]] = {},
supports_check_mode: bool = False,
add_file_common_args: bool = False,
) -> Dict[str, Any]: # pragma: nocover
return dict(
argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
required_together=required_together,
required_one_of=required_one_of,
required_if=required_if,
required_by=required_by,
add_file_common_args=add_file_common_args,
supports_check_mode=supports_check_mode,
)
def joindict(*items: dict) -> dict:
"""merges one or more dictionaries into one"""
odict = dict()
for item in items:
for key, value in item.items():
odict[key] = value
return odict