Commits vergleichen

..

Keine gemeinsamen Commits. „stable“ und „v0.3.1“ haben vollständig unterschiedliche Historien.

25 geänderte Dateien mit 181 neuen und 760 gelöschten Zeilen

1
.gitignore gevendort
Datei anzeigen

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

Datei anzeigen

@ -4,59 +4,10 @@ Sebastian.Base Release Notes
.. contents:: Topics .. contents:: Topics
v0.4.4
======
Minor Changes
-------------
- removed the empty options dict
v0.4.3
======
v0.4.2
======
Minor Changes
-------------
- removed the empty options dict
v0.4.1
======
Minor Changes
-------------
- added an default Display to the module
- fixed the docification of dictionaries
v0.4.0
======
Release Summary
---------------
to prevent empty sections, the install and header methods return None if the method would just the scetion
header
v0.3.1
======
Release Summary
---------------
removed forgotten print calls
v0.3.0 v0.3.0
====== ======
Release Summary
---------------
rewrote the Types helper
v0.2.0 v0.2.0
====== ======

Datei anzeigen

@ -1,13 +1,8 @@
VERSION := $(shell hatch version)
format: format:
black . black .
isort . isort .
version: changelog:
@yq --yaml-output-grammar-version 1.2 -i -y -s --arg 'version' "${VERSION}" '.[0] | (.version = $$version)' galaxy.yml
changelog: version
antsibull-changelog generate antsibull-changelog generate
docs: format docs: format
@ -16,11 +11,11 @@ docs: format
clean-dist: clean-dist:
rm -rf dist rm -rf dist
hatch-release: clean-dist version hatch-release:
hatch build hatch build
galaxy-release: clean-dist changelog version galaxy-release: clean-dist changelog
ansible-galaxy collection build --output-path dist/galaxy ansible-galaxy collection build --output-path dist
upload: galaxy-release hatch-release upload: galaxy-release hatch-release
./upload.sh ./upload.sh

Datei anzeigen

@ -16,4 +16,4 @@ plugins:
strategy: {} strategy: {}
test: {} test: {}
vars: {} vars: {}
version: 0.4.2 version: 0.2.0

Datei anzeigen

@ -9,51 +9,8 @@ releases:
0.3.0: 0.3.0:
changes: changes:
release_summary: rewrote the Types helper release_summary: rewrote the Types helper
fragments:
- 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:
- print_calls.yml
release_date: "2024-02-24" release_date: "2024-02-24"
0.4.0:
changes:
release_summary: "to prevent empty sections, the install and header methods return None if the method would just
the scetion
header
"
fragments:
- sectioning.yml
release_date: "2024-03-08"
0.4.1:
changes:
minor_changes:
- added an default Display to the module
- fixed the docification of dictionaries
fragments:
- display.yml
- options_fix.yml
release_date: "2024-03-13"
0.4.2:
changes:
minor_changes:
- removed the empty options dict
fragments:
- empty_options.yml
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

@ -3,7 +3,7 @@ changelog_filename_version_depth: 0
changes_file: changelog.yaml changes_file: changelog.yaml
changes_format: combined changes_format: combined
ignore_other_fragment_extensions: true ignore_other_fragment_extensions: true
keep_fragments: true keep_fragments: false
mention_ancestor: true mention_ancestor: true
new_plugins_after_name: removed_features new_plugins_after_name: removed_features
notesdir: fragments notesdir: fragments

Datei anzeigen

@ -1,6 +0,0 @@
---
minor_changes:
- Added an type hint to the update_doc function
- Added Type hint to the dictionary argument of the diff method
- added an wrapper for fail_json and exit_json
- improved the version update, the version is now only in one place the source of truth

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

@ -1,2 +0,0 @@
---
release_summary: change the module to an ansible module

Datei anzeigen

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

Datei anzeigen

@ -1,3 +0,0 @@
---
minor_changes:
- added an default Display to the module

Datei anzeigen

@ -1,3 +0,0 @@
---
minor_changes:
- removed the empty options dict

Datei anzeigen

@ -1,3 +0,0 @@
---
minor_changes:
- fixed the docification of dictionaries

Datei anzeigen

@ -1,2 +0,0 @@
---
release_summary: removed forgotten print calls

Datei anzeigen

@ -1,4 +0,0 @@
---
release_summary: |
to prevent empty sections, the install and header methods return None if the method would just the scetion
header

Datei anzeigen

@ -1,2 +0,0 @@
---
release_summary: rewrote the Types helper

Datei anzeigen

@ -1,22 +1,28 @@
---
namespace: sebastian namespace: sebastian
name: base name: base
version: 0.5.0 version: 0.3.1
readme: README.md readme: README.md
authors: authors:
- Sebastian Tobie - Sebastian Tobie
description: The base of my ansible collections. It provides the nessesary tools for description: >
my modules The base of my ansible collections. It provides the nessesary tools for my modules
license_file: LICENSE.txt license_file: LICENSE
tags: tags:
- linux - linux
- systemd - systemd
dependencies: {} dependencies: {}
repository: https://gitea.sebastian-tobie.de/ansible/ansible-module.git repository: https://gitea.sebastian-tobie.de/ansible/ansible-module.git
# documentation: https://gitea.sebastian-tobie.de/ansible/ansible-module
homepage: https://gitea.sebastian-tobie.de/ansible/ansible-module homepage: https://gitea.sebastian-tobie.de/ansible/ansible-module
issues: https://gitea.sebastian-tobie.de/ansible/ansible-module/issues issues: https://gitea.sebastian-tobie.de/ansible/ansible-module/issues
build_ignore: build_ignore:
- '*.gz' - "*.gz"
- .* - ".*"
- Makefile - Makefile
- pyproject.toml - pyproject.toml
- upload.sh - upload.sh
@ -24,5 +30,5 @@ build_ignore:
- changelogs - changelogs
- docs - docs
- src - src
- coverage.* - "coverage.*"
- dist - dist

Datei anzeigen

@ -1,71 +1,31 @@
import builtins
import pathlib import pathlib
import warnings import warnings
from typing import Any, Dict, Literal, NotRequired, Optional, Required, Sequence, Tuple, Type, TypedDict from typing import Any, Callable, Dict, Optional, Sequence, Tuple, Type, Union
__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"))
AnsibleType = Literal[ AnsibleParameter = Dict[str, Any]
"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 = {'__module__', '__name__', '__qualname__', '__doc__', '__annotations__', '__type_params__'}.difference( attrs = frozenset(
updates.keys() ('__module__', '__name__', '__qualname__', '__doc__', '__annotations__', '__type_params__')
) ) - frozenset(updates.keys())
for attr in attrs: for attr in attrs:
try: try:
value = getattr(func, attr) value = getattr(func, attr)
@ -83,14 +43,27 @@ 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."""
class TypeBase(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): 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",
@ -109,39 +82,17 @@ class TypeBase(type):
) )
for attr in types - set(attrs.keys()): for attr in types - set(attrs.keys()):
attrs[attr] = wrap_func( attrs[attr] = wrap_func(
default_type(attr), default(attr), __doc__=GENERIC_DOC.format(type=attr), __name__=attr, __qualname__=f"{clsname}.{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=TypeBase): class Types(metaclass=meta):
@staticmethod def list(
def _default(name: AnsibleType): elements: Union[Type[object], str, AnsibleParameter],
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]
elements: Type[object] | AnsibleType | AnsibleParameter,
help: str | list[str],
required: bool = False, required: bool = False,
default: list[Any] | None = None, help: Optional[str] = None,
) -> AnsibleParameter: ) -> AnsibleParameter:
"""Wrapper for the Ansible list type """Wrapper for the Ansible list type
@ -149,40 +100,26 @@ class Types(metaclass=TypeBase):
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.
""" """
if required and default: option: AnsibleParameter = dict(type="list", required=required)
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__ # type:ignore[reportGeneralTypeIssue] option["elements"] = elements.__name__
elif isinstance(elements, dict): elif isinstance(elements, dict):
option["elements"] = elements["type"] option["elements"] = elements["type"]
if elements["type"] == "dict" and "options" in elements: if elements["type"] == "dict":
option["options"] = dict() option["options"] = dict()
for name, value in elements["options"].items(): for name, value in elements["option"].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"
) )
elif "choices" in elements: if help is not None:
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
@staticmethod def dict(required: bool = False, help: Optional[str] = None, **options: AnsibleParameter) -> AnsibleParameter:
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:
@ -190,79 +127,14 @@ class Types(metaclass=TypeBase):
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 = AnsibleParameter(type="dict", description=help, required=required) option: AnsibleParameter = dict(type="dict", required=required)
option["options"] = options option["option"] = options
if help is not None and isinstance(help, str): if help is not None:
option["description"] = help.split("\n") option["description"] = help.split("\n")
elif help is not None:
option["description"] = help
return option return option
class ReturnTypes(metaclass=TypeBase): def systemdbool(b: Union[bool, str]) -> str:
@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"
@ -276,16 +148,9 @@ 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[Tuple[str, Any, Tuple[str, ...]] | Tuple[str, Any, Tuple[str, ...], bool]] = (), required_if: Sequence[Union[Tuple[str, Any, Tuple[str, ...]], Tuple[str, Any, Tuple[str, ...], bool]]] = (),
required_by: Dict[str, str | Tuple[str, ...]] = {}, required_by: Dict[str, Union[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,
@ -293,12 +158,6 @@ 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

@ -1,245 +0,0 @@
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,12 +1,11 @@
import os
import pathlib import pathlib
import shutil
from copy import deepcopy from copy import deepcopy
from typing import Any, Callable, ClassVar, Dict, Generic, NoReturn, Optional, Type, TypedDict, TypeVar, overload from typing import (Any, Callable, ClassVar, Dict, NoReturn, Optional, Type,
TypeVar, Union, overload)
import ansible.module_utils.basic as basic import ansible.module_utils.basic as basic
from .generic import AnsibleParameter, AnsibleReturnParameter, Types, modspec, systemdbool from .generic import AnsibleParameter, Types, systemdbool
__all__ = ( __all__ = (
"AnsibleModule", "AnsibleModule",
@ -19,35 +18,28 @@ __all__ = (
T = TypeVar("T") T = TypeVar("T")
class TypedDiff(Generic[T], TypedDict, total=False): def docify(input: Union[dict, AnsibleParameter]) -> dict:
before: T
after: T
before_header: str
after_header: str
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"]) # type: ignore[reportIndexIssue] options[name] = dict(type=help["type"])
if "description" in help: # type: ignore[reportOperatorIssue] if "description" in help:
if isinstance(help["description"], str): # type: ignore[reportIndexIssue] if isinstance(help["description"], str):
help["description"] = help["description"].split("\n") # type: ignore[reportIndexIssue] help["description"] = help["description"].split("\n")
options[name]["description"] = help["description"] # type: ignore[reportIndexIssue] options[name]["description"] = help["description"]
if "required" in help and help["required"]: # type: ignore[reportOperatorIssue] if "required" in help and help["required"]:
options[name]["required"] = True options[name]["required"] = True
else: else:
options[name]["required"] = False options[name]["required"] = False
if help["type"] == "list": # type: ignore[reportOperatorIssue] if help["type"] == "list":
options[name]["elements"] = help["elements"] # type: ignore[reportIndexIssue] options[name]["elements"] = help["elements"]
if not options[name]["required"]: if not options[name]["required"]:
options[name]["default"] = [] options[name]["default"] = []
if "default" in help: # type: ignore[reportOperatorIssue] if "default" in help:
options[name]["default"] = help["default"] # type: ignore[reportIndexIssue] options[name]["default"] = help["default"]
if "options" in help and help["options"] != {}: # type: ignore[reportOperatorIssue] if "options" in help:
options[name]["options"] = docify(help["options"]) # type: ignore[reportIndexIssue] options[name]["options"] = docify(help["options"])
if "choices" in help and len(help["choices"]) > 0: # type: ignore[reportOperatorIssue] if "choices" in help:
options[name]["choices"] = tuple(help["choices"]) # type: ignore[reportIndexIssue] options[name]["choices"] = tuple(help["choices"])
return options return options
@ -62,10 +54,6 @@ 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()
@ -74,11 +62,7 @@ class AnsibleModule(object):
"""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, documentation=False): def __init__(self):
"""
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))
@ -87,20 +71,15 @@ 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.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"""
self.result[key] = value self.result[key] = value
@overload @overload
def diff(self, diff: TypedDiff): # pragma: nocover def diff(self, diff: Dict[str, str]): # pragma: nocover
pass pass
@overload @overload
@ -128,7 +107,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 = TypedDiff( diff = dict(
before=before, before=before,
after=after, after=after,
) )
@ -179,8 +158,12 @@ class AnsibleModule(object):
except Exception as exc: except Exception as exc:
import traceback import traceback
self.fail("".join(traceback.format_exception(type(exc), exc, exc.__traceback__))) self.module.fail_json(
self.exit() "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)),
**self.result,
)
self.module.exit_json(**self.result)
raise Exception("exit_json failed")
@classmethod @classmethod
def doc(cls) -> str: def doc(cls) -> str:
@ -192,81 +175,26 @@ class AnsibleModule(object):
doc = cls.__doc__ doc = cls.__doc__
if doc is None: if doc is None:
doc = "" doc = ""
mod = cls(documentation=True) specs = dict()
options = docify(mod.modspec["argument_spec"]) if "argument_spec" in cls._common_args: # pragma: nocover
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(
documentation, dict(
module=cls.name,
short_description=docu[0],
description=docu,
options=options,
),
stream=None, stream=None,
explicit_start=True, explicit_start=True,
) )
) )
@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"""
self.module.fail_json(message, **self.result)
def exit(self) -> NoReturn: # type: ignore[reportReturnType]
"""Wrapper for AnsibleModule.exit_json"""
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
@ -301,37 +229,30 @@ class SystemdUnitModule(AnsibleModule):
), ),
) )
#: if defined it will be called after run has changed the unitfile #: if defined it will be called after run has changed the unitfile
post: Optional[Callable[[], None]] = None post: Optional[Callable[[], None]]
#: generates the install section if the unit is installable #: generates the install section if the unit is installable
install: ClassVar[Optional[Callable[["SystemdUnitModule"], str | None]]] = None install: ClassVar[Optional[Callable[["SystemdUnitModule"], str]]]
def unit(self) -> str: # pragma: nocover def unit(self) -> str: # pragma: nocover
raise NotImplementedError() raise NotImplementedError()
def header(self) -> Optional[str]: def header(self) -> str:
parts = self.map_param( header = "[Unit]\n"
description="Description", header += "".join(
documentation="Documentation", self.map_param(
requires="Requires", description="Description",
wants="Wants", documentation="Documentation",
partof="PartOf", requires="Requires",
before="Before", wants="Wants",
after="After", partof="PartOf",
before="Before",
after="After",
)
) )
if len(parts) == 0:
return None
header = "[Unit]\n" + "".join(parts)
return header return header
def _unit(self, *parts: Optional[str]) -> str: def map_param(self, **parammap: str):
opart = []
for part in parts:
if part is not None:
opart.append(part)
return "\n".join(opart)
def map_param(self, **parammap: str) -> list[str]:
"""maps an dict with keys for an section with given params. The key of the dictionary is the parameter and the value is the key in the unitfile. If an parameter has multiple values it adds multiple entries""" """maps an dict with keys for an section with given params. The key of the dictionary is the parameter and the value is the key in the unitfile. If an parameter has multiple values it adds multiple entries"""
output: list[str] = [] output: list[str] = []
for param, key in parammap.items(): for param, key in parammap.items():
@ -343,10 +264,38 @@ class SystemdUnitModule(AnsibleModule):
output.append(f"{key}={systemdbool(params)}\n") output.append(f"{key}={systemdbool(params)}\n")
return output return output
def unitfile_gen(self) -> None: # pragma: nocover def unitfile_gen(self): # pragma: nocover
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()
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())
@ -367,7 +316,10 @@ class SystemdUnitModule(AnsibleModule):
self.check() self.check()
if not self.changed: if not self.changed:
return return
self.move_file(self.tmpdir / "newunit", self.unitfile) self.module.atomic_move(
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()
@ -391,15 +343,13 @@ def installable(_class: Type[SystemdUnitModule]): # pragma: nocover
) )
specs["argument_spec"].update(arguments) specs["argument_spec"].update(arguments)
def install(self: SystemdUnitModule) -> Optional[str]: def install(self: SystemdUnitModule) -> str:
parts = [] output = "[Install]\n"
for argument, key in _INSTALL_MAPPING.items(): for argument, key in _INSTALL_MAPPING.items():
if self.get(argument, False): if self.get(argument, False):
for unit in self.get(argument): # type: ignore for unit in self.get(argument): # type: ignore
parts.append("{}={}\n".format(key, unit)) output += "{}={}\n".format(key, unit)
if len(parts) == 0: return output
return None
return "[Install]\n" + "".join(parts)
_class.install = install _class.install = install
_class.module_spec = specs _class.module_spec = specs

Datei anzeigen

@ -7,20 +7,23 @@ 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.11" requires-python = ">=3.7"
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 = ["ansible>=11.3.0"] dependencies = []
[tool.hatch.build]
directory = "dist/python"
[project.urls] [project.urls]
Documentation = "https://github.com/unknown/ansible-module#readme" Documentation = "https://github.com/unknown/ansible-module#readme"
Issues = "https://github.com/unknown/ansible-module/issues" Issues = "https://github.com/unknown/ansible-module/issues"
@ -143,11 +146,10 @@ exclude_lines = [
] ]
[tool.mypy] [tool.mypy]
disable_error_code = "import-untyped"
python_version = "3.11" python_version = "3.11"
warn_return_any = true warn_return_any = true
warn_unused_configs = true warn_unused_configs = true
[[tool.mypy.overrides]] [[tool.mypy.overrides]]
module = ["ansible", "ansible.module_utils", "ansible.module_utils.basic"] module = ["ansible", "ansible.module_utils", "ansible.module_utils.basic"]
ignore_missing_imports = true ignore_missing_imports = true

Datei anzeigen

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

Datei anzeigen

@ -4,18 +4,13 @@ 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( regex = re.compile("DOCUMENTATION *= *r?(?P<quote>\"{3}|'{3})(---)?.*?(?P=quote)", re.MULTILINE | re.DOTALL)
"(?P<type>DOCUMENTATION|RETURNS) *= *r?(?P<quote>\"{3}|'{3})(---)?.*?(?P=quote)",
re.MULTILINE | re.DOTALL,
)
def main() -> None: def main():
try: try:
modules = list(moduledir.iterdir()) modules = list(moduledir.iterdir())
except: except:
@ -25,7 +20,6 @@ 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__"):
@ -35,7 +29,6 @@ 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
@ -43,37 +36,16 @@ 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()
changed = False match = regex.search(moddata)
start = 0 if not match:
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(
modfile.write_text(moddata) pre=moddata[: match.start()], quote=match.group("quote"), doc=moddoc, post=moddata[match.end() :]
)
modfile.write_text(newmod)
print("updated the documentation of module {}".format(module.name)) print("updated the documentation of module {}".format(module.name))

Datei anzeigen

@ -3,5 +3,13 @@ user=$(yq -r .namespace galaxy.yml)
package=$(yq -r .name galaxy.yml) package=$(yq -r .name galaxy.yml)
version=$(yq -r .version galaxy.yml) version=$(yq -r .version galaxy.yml)
printf "Namespace: %s\nPackage: %s\nVersion: %s\n" $user $package $version printf "Namespace: %s\nPackage: %s\nVersion: %s\n" $user $package $version
ansible-galaxy collection publish dist/galaxy/* upload(){
hatch publish -r ansible dist/python/* name=$(basename "$1")
printf "uploading: %s as %s\n" "$1" "$name"
curl -u sebastian --upload-file "$1" "https://gitea.sebastian-tobie.de/api/packages/ansible/generic/${package}/${version}/$name"
}
for file in dist/sebastian* ; do
upload "$file"
done
rm -f dist/sebastian*
hatch publish -r ansible