Commits vergleichen

...

12 Commits

Autor SHA1 Nachricht Datum
3bb2a93dab updated the galaxy.yml 2025-03-22 16:35:18 +01:00
749ae48441 Release 0.5.0
- Added ReturnTypes
- improved the AnsibleParameter Type
- added an set containing  the arguments from add_file_common_args
- added an functio to move generated files to their location according to the common_args
- the modspec function was extended with arguments for metadata in the documentation
2025-03-22 16:29:19 +01:00
d82a902043 fixed an issue with choices in lists 2025-03-17 21:11:22 +01:00
328e58c439 improved the filemoving of AnsibleModules and SystemdUnitModule
I wrapped the file stuff in an function that is available in AnsibleModule.
2025-03-17 21:10:59 +01:00
9045e51c23 added ansible as dependency 2025-03-16 13:04:26 +01:00
74ded41d30 Release 0.4.4 2025-03-16 12:07:08 +01:00
59400d158f made some typehints that even work with pyright 2025-03-14 19:05:37 +01:00
144ffc72a5 some clean up of the code 2025-03-14 08:35:47 +01:00
c8d7d4e286 fixed an typing error 2025-03-14 08:35:28 +01:00
9d18ea533e removed display 2024-03-16 21:28:31 +01:00
f351bb6fb6 fixed typo 2024-03-16 21:28:18 +01:00
c5e6129682 modspec now supports supports_check_mode and add_file_common_args 2024-03-16 14:15:50 +01:00
13 geänderte Dateien mit 617 neuen und 151 gelöschten Zeilen

1
.gitignore gevendort
Datei anzeigen

@ -164,3 +164,4 @@ cython_debug/
# ---> Ansible # ---> Ansible
*.retry *.retry
src/ansible_module/module_utils

Datei anzeigen

@ -4,6 +4,16 @@ Sebastian.Base Release Notes
.. contents:: Topics .. contents:: Topics
v0.4.4
======
Minor Changes
-------------
- removed the empty options dict
v0.4.3
======
v0.4.2 v0.4.2
====== ======
@ -31,7 +41,6 @@ Release Summary
to prevent empty sections, the install and header methods return None if the method would just the scetion to prevent empty sections, the install and header methods return None if the method would just the scetion
header header
v0.3.1 v0.3.1
====== ======

Datei anzeigen

@ -4,44 +4,56 @@ releases:
changes: changes:
release_summary: change the module to an ansible module release_summary: change the module to an ansible module
fragments: fragments:
- base_release.yml - base_release.yml
release_date: '2024-02-11' release_date: "2024-02-11"
0.3.0: 0.3.0:
changes: changes:
release_summary: rewrote the Types helper release_summary: rewrote the Types helper
fragments: fragments:
- types.yml - types.yml
release_date: '2024-02-24' release_date: "2024-02-24"
0.3.1: 0.3.1:
changes: changes:
release_summary: removed forgotten print calls release_summary: removed forgotten print calls
fragments: fragments:
- print_calls.yml - print_calls.yml
release_date: '2024-02-24' release_date: "2024-02-24"
0.4.0: 0.4.0:
changes: changes:
release_summary: 'to prevent empty sections, the install and header methods release_summary: "to prevent empty sections, the install and header methods return None if the method would just
return None if the method would just the scetion the scetion
header header
' "
fragments: fragments:
- sectioning.yml - sectioning.yml
release_date: '2024-03-08' release_date: "2024-03-08"
0.4.1: 0.4.1:
changes: changes:
minor_changes: minor_changes:
- added an default Display to the module - added an default Display to the module
- fixed the docification of dictionaries - fixed the docification of dictionaries
fragments: fragments:
- display.yml - display.yml
- options_fix.yml - options_fix.yml
release_date: '2024-03-13' release_date: "2024-03-13"
0.4.2: 0.4.2:
changes: changes:
minor_changes: minor_changes:
- removed the empty options dict - removed the empty options dict
fragments: fragments:
- empty_options.yml - empty_options.yml
release_date: '2024-03-13' release_date: "2024-03-13"
0.4.3:
changes:
fragments:
- 0.4.3.yml
release_date: "2024-03-13"
0.4.4:
changes:
minor_changes:
- removed the empty options dict
fragments:
- 0.4.4.yml
release_date: "2025-03-16"

Datei anzeigen

@ -0,0 +1,3 @@
---
major_changes:
- the modules can now accept lists for help.

Datei anzeigen

@ -0,0 +1,2 @@
minor_changes:
- Added ansible as an dependency for the python module

Datei anzeigen

@ -0,0 +1,3 @@
---
minor_changes:
- modspec now supports supports_check_mode and add_file_common_args

Datei anzeigen

@ -1,6 +1,6 @@
namespace: sebastian namespace: sebastian
name: base name: base
version: 0.4.3 version: 0.5.0
readme: README.md readme: README.md
authors: authors:
- Sebastian Tobie - Sebastian Tobie

Datei anzeigen

@ -1,31 +1,71 @@
import builtins
import pathlib import pathlib
import warnings import warnings
from typing import Any, Callable, Dict, Optional, Sequence, Tuple, Type, Union from typing import Any, Dict, Literal, NotRequired, Optional, Required, Sequence, Tuple, Type, TypedDict
__all__ = ( __all__ = (
"Types", "Types",
"SYSTEMD_SERVICE_CONFIG", "SYSTEMD_SERVICE_CONFIG",
"SYSTEMD_NETWORK_CONFIG", "SYSTEMD_NETWORK_CONFIG",
"SYSTEMD_CONFIG_ROOT", "SYSTEMD_CONFIG_ROOT",
"FILE_COMMON_ARGS",
"systemdbool", "systemdbool",
"AnsibleParameter", "AnsibleParameter",
"AnsibleReturnParameter",
) )
SYSTEMD_CONFIG_ROOT = pathlib.Path("/etc/systemd") SYSTEMD_CONFIG_ROOT = pathlib.Path("/etc/systemd")
SYSTEMD_NETWORK_CONFIG = SYSTEMD_CONFIG_ROOT / "network" SYSTEMD_NETWORK_CONFIG = SYSTEMD_CONFIG_ROOT / "network"
SYSTEMD_SERVICE_CONFIG = SYSTEMD_CONFIG_ROOT / "system" SYSTEMD_SERVICE_CONFIG = SYSTEMD_CONFIG_ROOT / "system"
FILE_COMMON_ARGS = frozenset(("owner", "group", "mode", "seuser", "serole", "setype", "selevel", "unsafe_writes"))
AnsibleParameter = Dict[str, Any] AnsibleType = Literal[
"str",
"bool",
"int",
"float",
"path",
"raw",
"jsonarg",
"json",
"bytes",
"dict",
"list",
"bits",
]
ReturnOptions = Literal["always", "changed", "success"]
class AnsibleParameter(TypedDict, total=False):
description: Required[str | list[str]]
required: NotRequired[bool]
default: NotRequired[Any]
type: Required[AnsibleType]
choices: NotRequired[list[Any] | tuple[Any]]
elements: NotRequired["AnsibleParameter" | AnsibleType]
aliases: NotRequired[list[str]]
version_added: NotRequired[str]
options: NotRequired[dict[str, "AnsibleParameter"]]
class AnsibleReturnParameter(TypedDict, total=False):
description: Required[str | list[str]]
type: Required[AnsibleType]
returned: NotRequired[ReturnOptions]
elements: NotRequired[AnsibleType]
sample: NotRequired[list[Any] | Any]
version_added: NotRequired[str]
contains: NotRequired[dict[str, "AnsibleReturnParameter"]]
def wrap_func(func, **updates): def wrap_func(func, **updates):
def wrapper(*args, **kwargs): def wrapper(*args, **kwargs):
return func(*args, **kwargs) return func(*args, **kwargs)
attrs = frozenset( attrs = {'__module__', '__name__', '__qualname__', '__doc__', '__annotations__', '__type_params__'}.difference(
('__module__', '__name__', '__qualname__', '__doc__', '__annotations__', '__type_params__') updates.keys()
) - frozenset(updates.keys()) )
for attr in attrs: for attr in attrs:
try: try:
value = getattr(func, attr) value = getattr(func, attr)
@ -43,27 +83,14 @@ def wrap_func(func, **updates):
GENERIC_DOC = """Returns an dictionary for the Ansible {type} type.""" GENERIC_DOC = """Returns an dictionary for the Ansible {type} type."""
def default(name: str): class TypeBase(type):
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): def __new__(cls, clsname, bases, attrs):
if "_default" not in attrs:
raise TypeError(
f"The class {clsname} must define an wrapper function called _default that returns the default type"
)
default_type = attrs["_default"]
del attrs["_default"]
types = frozenset( types = frozenset(
( (
"str", "str",
@ -82,18 +109,39 @@ class meta(type):
) )
for attr in types - set(attrs.keys()): for attr in types - set(attrs.keys()):
attrs[attr] = wrap_func( attrs[attr] = wrap_func(
default(attr), __doc__=GENERIC_DOC.format(type=attr), __name__=attr, __qualname__=f"{clsname}.{attr}" default_type(attr),
__doc__=GENERIC_DOC.format(type=attr),
__name__=attr,
__qualname__=f"{clsname}.{attr}",
) )
attrs["__slots__"] = () attrs["__slots__"] = ()
return super().__new__(cls, clsname, bases, attrs) return super().__new__(cls, clsname, bases, attrs)
class Types(metaclass=meta): class Types(metaclass=TypeBase):
@staticmethod
def _default(name: AnsibleType):
def wrapped(
help: str | list[str],
required: bool = False,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
) -> AnsibleParameter:
option = AnsibleParameter(type=name, required=required, description=help)
if choices is not None:
option["choices"] = tuple(choices)
if default is not None:
option["default"] = default
return option
return wrapped
@staticmethod
def list( # type: ignore[misc] def list( # type: ignore[misc]
elements: Union[Type[object], str, AnsibleParameter], elements: Type[object] | AnsibleType | AnsibleParameter,
help: str | list[str],
required: bool = False, required: bool = False,
help: Optional[str] = None, default: list[Any] | None = None,
) -> AnsibleParameter: ) -> AnsibleParameter:
"""Wrapper for the Ansible list type """Wrapper for the Ansible list type
@ -101,26 +149,40 @@ class Types(metaclass=meta):
elements: The type of the elements elements: The type of the elements
required: if the item is absolutly required required: if the item is absolutly required
help: an helptext for the ansible-doc help: an helptext for the ansible-doc
default: an default value. The Value is not converted.
""" """
option: AnsibleParameter = dict(type="list", required=required) if required and default:
raise ValueError("required and default are not allowed")
option: AnsibleParameter = AnsibleParameter(
type="list",
required=required,
description=help,
)
if not isinstance(elements, (str, dict)): if not isinstance(elements, (str, dict)):
option["elements"] = elements.__name__ option["elements"] = elements.__name__ # type:ignore[reportGeneralTypeIssue]
elif isinstance(elements, dict): elif isinstance(elements, dict):
option["elements"] = elements["type"] option["elements"] = elements["type"]
if elements["type"] == "dict": if elements["type"] == "dict" and "options" in elements:
option["options"] = dict() option["options"] = dict()
for name, value in elements["option"].items(): for name, value in elements["options"].items():
option["options"][name] = value option["options"][name] = value
if "description" not in option["options"][name]: if "description" not in option["options"][name]:
warnings.warn( # pragma: nocover warnings.warn( # pragma: nocover
f"helptext of option {name} is unset." f"helptext of option {name} is unset."
" Ansible requires suboptions to have an documentation" " Ansible requires suboptions to have an documentation"
) )
if help is not None: elif "choices" in elements:
option["choices"] = elements["choices"]
if default is not None:
option["default"] = default
if help is not None and isinstance(help, str):
option["description"] = help.split("\n") option["description"] = help.split("\n")
elif help is not None:
option["description"] = help
return option return option
def dict(required: bool = False, help: Optional[str] = None, **options: AnsibleParameter) -> AnsibleParameter: # type: ignore[misc] @staticmethod
def dict(help: str | builtins.list[str], required: bool = False, **options: AnsibleParameter) -> AnsibleParameter: # type: ignore[misc]
"""Wrapper for the Ansible dict type """Wrapper for the Ansible dict type
Args: Args:
@ -128,14 +190,79 @@ class Types(metaclass=meta):
help: an helptext for the ansible-doc help: an helptext for the ansible-doc
options: The individual options that this parameter has options: The individual options that this parameter has
""" """
option: AnsibleParameter = dict(type="dict", required=required) option: AnsibleParameter = AnsibleParameter(type="dict", description=help, required=required)
option["options"] = options option["options"] = options
if help is not None: if help is not None and isinstance(help, str):
option["description"] = help.split("\n") option["description"] = help.split("\n")
elif help is not None:
option["description"] = help
return option return option
def systemdbool(b: Union[bool, str]) -> str: class ReturnTypes(metaclass=TypeBase):
@staticmethod
def _default(name: AnsibleType):
def wrapped(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
) -> AnsibleReturnParameter:
option = AnsibleReturnParameter(type=name, description=help)
if returned is not None:
option["returned"] = returned
if sample is not None:
option["sample"] = sample
if version_added is not None:
option["version_added"] = version_added
return option
return wrapped
@staticmethod
def list(
help: str | list[str],
elements: AnsibleType | AnsibleReturnParameter | Type[object],
returned: ReturnOptions | None = None,
sample: builtins.list[Any] | None = None,
version_added: str | None = None,
) -> AnsibleReturnParameter:
option = AnsibleReturnParameter(description=help, type="list")
if returned is not None:
option["returned"] = returned
if isinstance(elements, str):
option["elements"] = elements
elif isinstance(elements, dict):
option["elements"] = elements["type"]
if "contains" in elements:
option["contains"] = elements["contains"]
else:
option["elements"] = elements.__name__ # type:ignore[reportGeneralTypeIssue]
if sample is not None:
option["sample"] = sample
if version_added is not None:
option["version_added"] = version_added
return option
@staticmethod
def dict(
help: str | builtins.list[str],
contains: dict[str, AnsibleReturnParameter],
returned: ReturnOptions | None = None,
sample: builtins.list[Any] | None = None,
version_added: str | None = None,
):
option = AnsibleReturnParameter(description=help, type="dict", contains=contains)
if returned is not None:
option["contains"] = contains
if sample is not None:
option["sample"] = sample
if version_added is not None:
option["version_added"] = version_added
return option
def systemdbool(b: bool | str) -> str:
"""Converts values into things systemd can parse""" """Converts values into things systemd can parse"""
if b is True: if b is True:
return "yes" return "yes"
@ -149,9 +276,16 @@ def modspec(
mutually_exclusive: Sequence[Tuple[str, ...]] = (), mutually_exclusive: Sequence[Tuple[str, ...]] = (),
required_together: Sequence[Tuple[str, ...]] = (), required_together: Sequence[Tuple[str, ...]] = (),
required_one_of: 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_if: Sequence[Tuple[str, Any, Tuple[str, ...]] | Tuple[str, Any, Tuple[str, ...], bool]] = (),
required_by: Dict[str, Union[str, Tuple[str, ...]]] = {}, required_by: Dict[str, str | Tuple[str, ...]] = {},
supports_check_mode: bool = False,
add_file_common_args: bool = False,
deprecated: bool = False,
version_added: str | None = None,
notes: list[str] | None = None,
extends_documentation_fragment: list[str] | None = None,
) -> Dict[str, Any]: # pragma: nocover ) -> Dict[str, Any]: # pragma: nocover
"""Wrapper to properly Type the module specs"""
return dict( return dict(
argument_spec=argument_spec, argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive, mutually_exclusive=mutually_exclusive,
@ -159,6 +293,12 @@ def modspec(
required_one_of=required_one_of, required_one_of=required_one_of,
required_if=required_if, required_if=required_if,
required_by=required_by, required_by=required_by,
add_file_common_args=add_file_common_args,
supports_check_mode=supports_check_mode,
deprecated=deprecated,
version_added=version_added,
notes=notes,
extends_documentation_fragment=extends_documentation_fragment,
) )

Datei anzeigen

@ -0,0 +1,245 @@
from pathlib import PosixPath
from typing import Any, Dict, Literal, NotRequired, Optional, Required, Sequence, Tuple, Type, TypedDict
__all__ = [
"Types",
"SYSTEMD_SERVICE_CONFIG",
"SYSTEMD_NETWORK_CONFIG",
"SYSTEMD_CONFIG_ROOT",
"FILE_COMMON_ARGS",
"systemdbool",
"AnsibleParameter",
"AnsibleReturnParameter",
]
SYSTEMD_CONFIG_ROOT: PosixPath
SYSTEMD_NETWORK_CONFIG: PosixPath
SYSTEMD_SERVICE_CONFIG: PosixPath
FILE_COMMON_ARGS: frozenset[str]
AnsibleType = Literal[
"str",
"bool",
"int",
"float",
"path",
"raw",
"jsonarg",
"json",
"bytes",
"dict",
"list",
"bits",
]
ReturnOptions = Literal["always", "changed", "success"]
class AnsibleParameter(TypedDict, total=False):
description: Required[str | list[str]]
required: NotRequired[bool]
default: NotRequired[Any]
type: Required[AnsibleType]
choices: NotRequired[list[Any] | tuple[Any]]
elements: NotRequired["AnsibleParameter" | AnsibleType]
aliases: NotRequired[list[str]]
version_added: NotRequired[str]
options: NotRequired[dict[str, "AnsibleParameter"]]
class AnsibleReturnParameter(TypedDict, total=False):
description: Required[str | list[str]]
type: Required[AnsibleType]
returned: NotRequired[ReturnOptions]
elements: NotRequired[AnsibleType]
sample: NotRequired[list[Any] | Any]
version_added: NotRequired[str]
contains: NotRequired[dict[str, "AnsibleReturnParameter"]]
class TypeBase(type):
def __new__(cls, clsname, bases, attrs): ...
class Types(metaclass=TypeBase):
@staticmethod
def str(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def bool(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def int(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def float(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def path(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def raw(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def jsonarg(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def json(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def bytes(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def bits(
required: bool = False,
help: str | list[str] | None = None,
choices: Optional[Sequence] = None,
default: Optional[Any] = None,
): ...
@staticmethod
def list(
elements: type[object] | str | AnsibleParameter,
required: bool = False,
help: str | list[str] | None = None,
default: list[Any] | None = None,
) -> AnsibleParameter: ...
@staticmethod
def dict(
required: bool = False, help: str | list[str] | None = None, **options: AnsibleParameter
) -> AnsibleParameter: ...
class ReturnTypes(metaclass=TypeBase):
@staticmethod
def list(
help: str | list[str],
elements: AnsibleType | AnsibleReturnParameter | Type[object],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def dict(
help: str | list[str],
contains: dict[str, AnsibleReturnParameter],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def str(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def bool(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def int(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def float(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def path(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def raw(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def jsonarg(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def json(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def bytes(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
@staticmethod
def bits(
help: str | list[str],
returned: ReturnOptions | None = None,
sample: list[Any] | None = None,
version_added: str | None = None,
): ...
def systemdbool(b: bool | str) -> str: ...
def joindict(*items: dict) -> dict: ...
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[Tuple[str, Any, Tuple[str, ...]] | Tuple[str, Any, Tuple[str, ...], bool]] = (),
required_by: Dict[str, str | Tuple[str, ...]] = {},
supports_check_mode: bool = False,
add_file_common_args: bool = False,
deprecated: bool = False,
version_added: str | None = None,
notes: list[str] | None = None,
extends_documentation_fragment: list[str] | None = None,
) -> Dict[str, Any]: ...

Datei anzeigen

@ -1,11 +1,12 @@
import os
import pathlib import pathlib
import shutil
from copy import deepcopy from copy import deepcopy
from typing import Any, Callable, ClassVar, Dict, NoReturn, Optional, Type, TypeVar, Union, overload, TypedDict from typing import Any, Callable, ClassVar, Dict, Generic, NoReturn, Optional, Type, TypedDict, TypeVar, overload
from ansible.utils.display import Display
import ansible.module_utils.basic as basic import ansible.module_utils.basic as basic
from .generic import AnsibleParameter, Types, systemdbool from .generic import AnsibleParameter, AnsibleReturnParameter, Types, modspec, systemdbool
__all__ = ( __all__ = (
"AnsibleModule", "AnsibleModule",
@ -18,35 +19,35 @@ __all__ = (
T = TypeVar("T") T = TypeVar("T")
class TypedDiff(TypedDict): class TypedDiff(Generic[T], TypedDict, total=False):
before: str before: T
after: str after: T
before_header: Optional[str] = None before_header: str
after_header: Optional[str] = None after_header: str
def docify(input: Union[dict, AnsibleParameter]) -> dict: def docify(input: AnsibleParameter) -> dict:
options = dict() options = dict()
for name, help in input.items(): for name, help in input.items():
options[name] = dict(type=help["type"]) options[name] = dict(type=help["type"]) # type: ignore[reportIndexIssue]
if "description" in help: if "description" in help: # type: ignore[reportOperatorIssue]
if isinstance(help["description"], str): if isinstance(help["description"], str): # type: ignore[reportIndexIssue]
help["description"] = help["description"].split("\n") help["description"] = help["description"].split("\n") # type: ignore[reportIndexIssue]
options[name]["description"] = help["description"] options[name]["description"] = help["description"] # type: ignore[reportIndexIssue]
if "required" in help and help["required"]: if "required" in help and help["required"]: # type: ignore[reportOperatorIssue]
options[name]["required"] = True options[name]["required"] = True
else: else:
options[name]["required"] = False options[name]["required"] = False
if help["type"] == "list": if help["type"] == "list": # type: ignore[reportOperatorIssue]
options[name]["elements"] = help["elements"] options[name]["elements"] = help["elements"] # type: ignore[reportIndexIssue]
if not options[name]["required"]: if not options[name]["required"]:
options[name]["default"] = [] options[name]["default"] = []
if "default" in help: if "default" in help: # type: ignore[reportOperatorIssue]
options[name]["default"] = help["default"] options[name]["default"] = help["default"] # type: ignore[reportIndexIssue]
if "options" in help and help["options"] != {}: if "options" in help and help["options"] != {}: # type: ignore[reportOperatorIssue]
options[name]["options"] = docify(help["options"]) options[name]["options"] = docify(help["options"]) # type: ignore[reportIndexIssue]
if "choices" in help and len(help["choices"]) > 0: if "choices" in help and len(help["choices"]) > 0: # type: ignore[reportOperatorIssue]
options[name]["choices"] = tuple(help["choices"]) options[name]["choices"] = tuple(help["choices"]) # type: ignore[reportIndexIssue]
return options return options
@ -61,17 +62,23 @@ class AnsibleModule(object):
result: dict result: dict
#: the specification of the arguments. Subclasses that are usable Modules must set this value. #: the specification of the arguments. Subclasses that are usable Modules must set this value.
module_spec: ClassVar[dict] module_spec: ClassVar[dict]
#: The specification of return arguments
return_spec: ClassVar[dict[str, AnsibleReturnParameter]]
#: This is set by classes that define common things for their subclasses, like behaviour of the run and check methods. This is used by `SystemdUnitModule` #: This is set by classes that define common things for their subclasses, like behaviour of the run and check methods. This is used by `SystemdUnitModule`
_common_args: ClassVar[dict[str, Any]] = dict() _common_args: ClassVar[dict[str, Any]] = dict()
#: For log messages
display: Display
@property @property
def params(self) -> Dict[str, Any]: def params(self) -> Dict[str, Any]:
"""params is an wrapper for the module.params""" """params is an wrapper for the module.params"""
return self.module.params # type: ignore return self.module.params # type: ignore
def __init__(self): def __init__(self, documentation=False):
"""
Args:
documentation: Only true if the module is initialized for documentation. The Ansible Module is None is this case
"""
self.result = dict(changed=False) self.result = dict(changed=False)
specs = dict() specs = dict()
specs.update(deepcopy(self._common_args)) specs.update(deepcopy(self._common_args))
@ -80,9 +87,13 @@ class AnsibleModule(object):
specs["argument_spec"].update(modspec["argument_spec"]) specs["argument_spec"].update(modspec["argument_spec"])
del modspec["argument_spec"] del modspec["argument_spec"]
specs.update(modspec) specs.update(modspec)
self.display = Display() self.modspec = specs.copy()
self.module = basic.AnsibleModule(**specs) specs.pop("deprecated", None)
self.tmpdir = pathlib.Path(self.module.tmpdir) specs.pop("notes", None)
specs.pop("version_added", None)
if not documentation:
self.module = basic.AnsibleModule(**specs)
self.tmpdir = pathlib.Path(self.module.tmpdir)
def set(self, key: str, value): def set(self, key: str, value):
"""sets an value for the result""" """sets an value for the result"""
@ -117,7 +128,7 @@ class AnsibleModule(object):
if diff is not None and not any((before is not None, after is not None)): if diff is not None and not any((before is not None, after is not None)):
pass pass
elif all((before is not None, after is not None, diff is None)): elif all((before is not None, after is not None, diff is None)):
diff = dict( diff = TypedDiff(
before=before, before=before,
after=after, after=after,
) )
@ -181,34 +192,81 @@ class AnsibleModule(object):
doc = cls.__doc__ doc = cls.__doc__
if doc is None: if doc is None:
doc = "" doc = ""
specs = dict() mod = cls(documentation=True)
if "argument_spec" in cls._common_args: # pragma: nocover options = docify(mod.modspec["argument_spec"])
specs.update(cls._common_args["argument_spec"])
if "argument_spec" in cls.module_spec: # pragma: nocover
specs.update(cls.module_spec["argument_spec"])
options = docify(specs)
docu = doc.split("\n") docu = doc.split("\n")
documentation: dict[str, Any] = dict(
module=cls.name,
short_description=docu[0],
description=docu,
options=options,
)
if mod.modspec.get("extends_documentation_fragment", None) is not None or mod.modspec.get(
"add_file_common_args", False
):
documentation["extends_documentation_fragment"] = []
if mod.modspec.get("extends_documentation_fragment", None) is not None:
documentation["extends_documentation_fragment"].extend(mod.modspec["extends_documentation_fragment"])
if mod.modspec.get("add_file_common_args", False):
documentation["extends_documentation_fragment"].append("ansible.builtin.files")
if mod.modspec.get("deprecated", False):
documentation["deprecated"] = True
if mod.modspec.get("notes", None) is not None:
documentation["notes"] = mod.modspec["notes"]
if mod.modspec.get("version_added", None) is not None:
documentation["version_added"] = mod.modspec["version_added"]
return str( return str(
yaml.safe_dump( yaml.safe_dump(
dict( documentation,
module=cls.name,
short_description=docu[0],
description=docu,
options=options,
),
stream=None, stream=None,
explicit_start=True, explicit_start=True,
) )
) )
def fail(self, message: str) -> NoReturn: @classmethod
def returns(cls) -> str:
"""This returns the string for the RETURNS String of the module documentation"""
try:
import yaml
except ImportError: # pragma: nocover
return "---\n"
if not hasattr(cls, "return_spec"):
return "--\n"
return str(
yaml.safe_dump(
cls.returns,
stream=None,
explicit_start=True,
)
)
def fail(self, message: str) -> NoReturn: # type: ignore[reportReturnType]
"""Wrapper for AnsibleModule.fail_json""" """Wrapper for AnsibleModule.fail_json"""
self.module.fail_json(message, **self.result) self.module.fail_json(message, **self.result)
def exit(self) -> NoReturn: def exit(self) -> NoReturn: # type: ignore[reportReturnType]
"""Wrapper for AnsibleModule.exit_json""" """Wrapper for AnsibleModule.exit_json"""
self.module.exit_json(**self.result) self.module.exit_json(**self.result)
def move_file(self, path: pathlib.Path, dest: pathlib.Path, backup: bool = False, unsafe_writes: bool = False):
"""Moves an Temporary file to an destination it uses the args from add_file_common_args when used
Args:
path: The Path that the file currently is
dest: the location the file should be
backup: should an backup be made before the file is moved
"""
if backup:
shutil.copy2(dest, dest.with_suffix(dest.suffix + ".bak"), follow_symlinks=False)
self.module.atomic_move(path, dest, unsafe_writes=unsafe_writes, keep_dest_attrs=True)
if "add_file_common_args" in self.modspec and self.modspec["add_file_common_args"]:
file_args = self.module.load_file_common_arguments(self.params, path=dest)
diff = TypedDiff()
self.changed |= self.module.set_fs_attributes_if_different(file_args, diff=diff)
self.diff(diff)
class SystemdUnitModule(AnsibleModule): class SystemdUnitModule(AnsibleModule):
#: path of the unitfile managed by this module #: path of the unitfile managed by this module
@ -289,34 +347,6 @@ class SystemdUnitModule(AnsibleModule):
path = self.tmpdir / "newunit" path = self.tmpdir / "newunit"
with open(path, "w") as unit: with open(path, "w") as unit:
unit.write(self.unit()) unit.write(self.unit())
self.module.set_owner_if_different(path.as_posix(), "root", False)
self.module.set_group_if_different(path.as_posix(), "root", False)
self.module.set_mode_if_different(path.as_posix(), "0644", False)
if self.unitfile.exists():
diff: dict[str, str] = dict()
self.changed = self.changed | self.module.set_owner_if_different(
self.unitfile.as_posix(),
"root",
self.result["changed"],
diff,
)
self.diff(diff)
diff = dict()
self.changed = self.changed | self.module.set_group_if_different(
self.unitfile.as_posix(),
"root",
self.result["changed"],
diff,
)
self.diff(diff)
diff = dict()
self.changed = self.changed | self.module.set_mode_if_different(
self.unitfile.as_posix(),
"0644",
self.result["changed"],
diff,
)
self.diff(diff)
def check(self): # pragma: nocover def check(self): # pragma: nocover
self.set("unitfile", self.unitfile.as_posix()) self.set("unitfile", self.unitfile.as_posix())
@ -337,10 +367,7 @@ class SystemdUnitModule(AnsibleModule):
self.check() self.check()
if not self.changed: if not self.changed:
return return
self.module.atomic_move( self.move_file(self.tmpdir / "newunit", self.unitfile)
src=(self.tmpdir / "newunit").as_posix(),
dest=self.unitfile.as_posix(),
)
if hasattr(self, "post") and self.post is not None: if hasattr(self, "post") and self.post is not None:
self.post() self.post()

Datei anzeigen

@ -7,22 +7,18 @@ name = "ansible-module"
dynamic = ["version"] dynamic = ["version"]
description = 'Helps with developing modules for ansible in an easier manner' description = 'Helps with developing modules for ansible in an easier manner'
readme = "README.md" readme = "README.md"
requires-python = ">=3.7" requires-python = ">=3.11"
license = "MIT" license = "MIT"
keywords = [] keywords = []
authors = [{ name = "Sebastian Tobie", email = "sebastian@sebastian-tobie.de" }] authors = [{ name = "Sebastian Tobie", email = "sebastian@sebastian-tobie.de" }]
classifiers = [ classifiers = [
"Development Status :: 4 - Beta", "Development Status :: 4 - Beta",
"Programming Language :: Python", "Programming Language :: Python",
"Programming Language :: Python :: 3.7",
"Programming Language :: Python :: 3.8",
"Programming Language :: Python :: 3.9",
"Programming Language :: Python :: 3.10",
"Programming Language :: Python :: 3.11", "Programming Language :: Python :: 3.11",
"Programming Language :: Python :: Implementation :: CPython", "Programming Language :: Python :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy", "Programming Language :: Python :: Implementation :: PyPy",
] ]
dependencies = [] dependencies = ["ansible>=11.3.0"]
[tool.hatch.build] [tool.hatch.build]
directory = "dist/python" directory = "dist/python"
[project.urls] [project.urls]

Datei anzeigen

@ -1 +1 @@
__version__ = "0.4.3" __version__ = "0.5.0"

Datei anzeigen

@ -4,10 +4,15 @@ import pathlib
import re import re
import sys import sys
from .module_utils.module import AnsibleModule
sys.path.append(".") sys.path.append(".")
mindocstring = "DOCUMENTATION = ''''''" mindocstring = "DOCUMENTATION = ''''''"
moduledir = pathlib.Path("plugins/modules") moduledir = pathlib.Path("plugins/modules")
regex = re.compile("DOCUMENTATION *= *r?(?P<quote>\"{3}|'{3})(---)?.*?(?P=quote)", re.MULTILINE | re.DOTALL) regex = re.compile(
"(?P<type>DOCUMENTATION|RETURNS) *= *r?(?P<quote>\"{3}|'{3})(---)?.*?(?P=quote)",
re.MULTILINE | re.DOTALL,
)
def main() -> None: def main() -> None:
@ -20,6 +25,7 @@ def main() -> None:
if modfile.name in ("__init__.py", "__pycache__", "unit.py.example"): if modfile.name in ("__init__.py", "__pycache__", "unit.py.example"):
continue continue
mod = importlib.import_module(".".join((modfile.parts[:-1]) + (modfile.stem,))) mod = importlib.import_module(".".join((modfile.parts[:-1]) + (modfile.stem,)))
module: AnsibleModule
if hasattr(mod, "Module"): if hasattr(mod, "Module"):
module = mod.Module module = mod.Module
elif hasattr(mod, "__module_name__"): elif hasattr(mod, "__module_name__"):
@ -29,6 +35,7 @@ def main() -> None:
continue continue
try: try:
moddoc = module.doc() moddoc = module.doc()
returns = module.returns()
except AttributeError: except AttributeError:
print("Broken module. skipping {}".format(modfile)) print("Broken module. skipping {}".format(modfile))
continue continue
@ -36,16 +43,37 @@ def main() -> None:
print("Error in documentation of module {}: {}".format(modfile, e)) print("Error in documentation of module {}: {}".format(modfile, e))
continue continue
moddata = modfile.read_text() moddata = modfile.read_text()
match = regex.search(moddata) changed = False
if not match: start = 0
while True:
match = regex.search(moddata, pos=start)
if not match:
break
changed = True
type = match.group("type")
content = ""
if type == "DOCUMENTATION":
content = moddoc
elif type == "RETURNS":
content = returns
else:
Exception("Please update the script: Unknown Type")
after = moddata[match.end() :]
moddata = "{pre}{type} = r{quote}{doc}{quote}".format(
pre=moddata[: match.start()],
type=type,
quote=match.group("quote"),
doc=content,
)
start = len(moddata)
moddata = f"{moddata}{after}"
if changed is False:
print( print(
"no Documentation set for module {}. Please add at least \"{}\" to the file".format( "no Documentation set for module {}. Please add at least \"{}\" to the file".format(
modfile.stem, mindocstring modfile.stem, mindocstring
) )
) )
continue continue
newmod = "{pre}DOCUMENTATION = {quote}{doc}{quote}{post}".format(
pre=moddata[: match.start()], quote=match.group("quote"), doc=moddoc, post=moddata[match.end() :] modfile.write_text(moddata)
)
modfile.write_text(newmod)
print("updated the documentation of module {}".format(module.name)) print("updated the documentation of module {}".format(module.name))