2023-11-25 11:16:52 +00:00
|
|
|
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]
|
|
|
|
|
|
|
|
|
2024-02-24 09:20:47 +00:00
|
|
|
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:
|
|
|
|
print(f"changing:\n {value}\n {getattr(wrapper, attr)}")
|
|
|
|
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
|
2023-11-25 11:16:52 +00:00
|
|
|
|
2024-02-24 09:20:47 +00:00
|
|
|
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()):
|
|
|
|
print(f"adding {attr} function")
|
|
|
|
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)
|
2023-11-25 11:16:52 +00:00
|
|
|
|
|
|
|
|
2024-02-24 09:20:47 +00:00
|
|
|
class Types(metaclass=meta):
|
2023-11-25 11:16:52 +00:00
|
|
|
def list(
|
|
|
|
elements: Union[Type[object], str, AnsibleParameter],
|
|
|
|
required: bool = False,
|
|
|
|
help: Optional[str] = None,
|
|
|
|
) -> AnsibleParameter:
|
2024-02-24 09:20:47 +00:00
|
|
|
"""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
|
|
|
|
"""
|
2023-11-25 11:16:52 +00:00
|
|
|
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
|
|
|
|
|
2024-02-24 09:20:47 +00:00
|
|
|
def dict(required: bool = False, help: Optional[str] = None, **options: AnsibleParameter) -> AnsibleParameter:
|
|
|
|
"""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
|
|
|
|
"""
|
2023-11-25 11:16:52 +00:00
|
|
|
option: AnsibleParameter = dict(type="dict", required=required)
|
|
|
|
option["option"] = 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, ...]]] = {},
|
|
|
|
) -> 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,
|
|
|
|
)
|
|
|
|
|
|
|
|
|
|
|
|
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
|