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["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