Commits vergleichen

...

26 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
1ea397da14 improved the build process 2024-03-16 10:39:59 +01:00
84eb37ffa7 added wrapper for fail_json and exit_json 2024-03-16 10:04:38 +01:00
85880d2867 added type hint for the diff. 2024-03-16 09:56:39 +01:00
380bc582e3 added returntype 2024-03-16 09:28:26 +01:00
8190ee94d3 fixed the empty options 2024-03-13 20:17:36 +01:00
2f36e69a73 fixed the documentation for dictionaries 2024-03-13 20:13:35 +01:00
1da65334e2 added an default display 2024-03-12 21:18:09 +01:00
47c8d3319e added fragments again 2024-03-12 21:10:44 +01:00
cd1d5aa84e improved the build and upload 2024-03-08 22:44:08 +01:00
0ba608406e to prevent empty sections, the install and header methods return None if the method would just the scetion header 2024-03-08 21:58:32 +01:00
d03b45d678 updated changelog 2024-03-04 23:06:53 +01:00
e583ad0f96 small remove print calls 2024-02-24 10:45:49 +01:00
8dbb67faf8 the upload script now upload python and ansible packages correctly 2024-02-24 10:21:48 +01:00
e089dbc82a rewrote the Types class 2024-02-24 10:20:47 +01:00
25 geänderte Dateien mit 825 neuen und 198 gelöschten Zeilen

1
.gitignore gevendort
Datei anzeigen

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

Datei anzeigen

@ -4,6 +4,58 @@ Sebastian.Base Release Notes
.. 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
======
Release Summary
---------------
rewrote the Types helper
v0.2.0
======

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

@ -4,5 +4,56 @@ releases:
changes:
release_summary: change the module to an ansible module
fragments:
- base_release.yml
release_date: '2024-02-11'
- base_release.yml
release_date: "2024-02-11"
0.3.0:
changes:
release_summary: rewrote the Types helper
fragments:
- types.yml
release_date: "2024-02-24"
0.3.1:
changes:
release_summary: removed forgotten print calls
fragments:
- print_calls.yml
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_format: combined
ignore_other_fragment_extensions: true
keep_fragments: false
keep_fragments: true
mention_ancestor: true
new_plugins_after_name: removed_features
notesdir: fragments

Datei anzeigen

@ -0,0 +1,6 @@
---
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

@ -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,2 @@
---
release_summary: change the module to an ansible module

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

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

Datei anzeigen

@ -1,105 +1,268 @@
import builtins
import pathlib
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__ = (
"Types",
"SYSTEMD_SERVICE_CONFIG",
"SYSTEMD_NETWORK_CONFIG",
"SYSTEMD_CONFIG_ROOT",
"FILE_COMMON_ARGS",
"systemdbool",
"AnsibleParameter",
"AnsibleReturnParameter",
)
SYSTEMD_CONFIG_ROOT = pathlib.Path("/etc/systemd")
SYSTEMD_NETWORK_CONFIG = SYSTEMD_CONFIG_ROOT / "network"
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 _Type:
def __dir__(self) -> tuple: # pragma: nocover
return (
"str",
"bool",
"int",
"float",
"path",
"raw",
"jsonarg",
"json",
"bytes",
"dict",
"list",
"bits",
"__doc__",
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 wrapper(*args, **kwargs):
return func(*args, **kwargs)
attrs = {'__module__', '__name__', '__qualname__', '__doc__', '__annotations__', '__type_params__'}.difference(
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."""
class TypeBase(type):
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(
(
"str",
"bool",
"int",
"float",
"path",
"raw",
"jsonarg",
"json",
"bytes",
"dict",
"list",
"bits",
)
)
for attr in types - set(attrs.keys()):
attrs[attr] = wrap_func(
default_type(attr),
__doc__=GENERIC_DOC.format(type=attr),
__name__=attr,
__qualname__=f"{clsname}.{attr}",
)
attrs["__slots__"] = ()
return super().__new__(cls, clsname, bases, attrs)
def __repr__(self) -> str: # pragma: nocover
return "Types()"
def __call__(self) -> "_Type": # pragma: nocover
return self
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
def list(
self,
elements: Union[Type[object], str, AnsibleParameter],
return wrapped
@staticmethod
def list( # type: ignore[misc]
elements: Type[object] | AnsibleType | AnsibleParameter,
help: str | list[str],
required: bool = False,
help: Optional[str] = None,
default: list[Any] | None = None,
) -> AnsibleParameter:
option: AnsibleParameter = dict(type="list", required=required)
"""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
default: an default value. The Value is not converted.
"""
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)):
option["elements"] = elements.__name__
option["elements"] = elements.__name__ # type:ignore[reportGeneralTypeIssue]
elif isinstance(elements, dict):
option["elements"] = elements["type"]
if elements["type"] == "dict":
if elements["type"] == "dict" and "options" in elements:
option["options"] = dict()
for name, value in elements["option"].items():
for name, value in elements["options"].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:
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")
elif help is not None:
option["description"] = help
return option
def dict(self, required: bool = False, help: Optional[str] = None, **options: dict) -> AnsibleParameter:
option: AnsibleParameter = dict(type="dict", required=required)
option["option"] = options
if help is not None:
@staticmethod
def dict(help: str | builtins.list[str], required: bool = False, **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 = AnsibleParameter(type="dict", description=help, required=required)
option["options"] = options
if help is not None and isinstance(help, str):
option["description"] = help.split("\n")
elif help is not None:
option["description"] = help
return option
def __getattr__(self, name: str):
def argument(
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")
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
argument.__name__ = name
argument.__qualname__ = f"Types.{name}"
argument.__doc__ = f"Simple wrapper for Ansible {name} argument dict"
return argument
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
Types = _Type()
def systemdbool(b: Union[bool, str]) -> str:
def systemdbool(b: bool | str) -> str:
"""Converts values into things systemd can parse"""
if b is True:
return "yes"
@ -113,9 +276,16 @@ def modspec(
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, ...]]] = {},
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]: # pragma: nocover
"""Wrapper to properly Type the module specs"""
return dict(
argument_spec=argument_spec,
mutually_exclusive=mutually_exclusive,
@ -123,6 +293,12 @@ def modspec(
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,
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 shutil
from copy import deepcopy
from typing import (Any, Callable, ClassVar, Dict, NoReturn, Optional, Type,
TypeVar, Union, overload)
from typing import Any, Callable, ClassVar, Dict, Generic, NoReturn, Optional, Type, TypedDict, TypeVar, overload
import ansible.module_utils.basic as basic
from .generic import AnsibleParameter, Types, systemdbool
from .generic import AnsibleParameter, AnsibleReturnParameter, Types, modspec, systemdbool
__all__ = (
"AnsibleModule",
@ -18,28 +19,35 @@ __all__ = (
T = TypeVar("T")
def docify(input: Union[dict, AnsibleParameter]) -> dict:
class TypedDiff(Generic[T], TypedDict, total=False):
before: T
after: T
before_header: str
after_header: str
def docify(input: AnsibleParameter) -> dict:
options = dict()
for name, help in input.items():
options[name] = dict(type=help["type"])
if "description" in help:
if isinstance(help["description"], str):
help["description"] = help["description"].split("\n")
options[name]["description"] = help["description"]
if "required" in help and help["required"]:
options[name] = dict(type=help["type"]) # type: ignore[reportIndexIssue]
if "description" in help: # type: ignore[reportOperatorIssue]
if isinstance(help["description"], str): # type: ignore[reportIndexIssue]
help["description"] = help["description"].split("\n") # type: ignore[reportIndexIssue]
options[name]["description"] = help["description"] # type: ignore[reportIndexIssue]
if "required" in help and help["required"]: # type: ignore[reportOperatorIssue]
options[name]["required"] = True
else:
options[name]["required"] = False
if help["type"] == "list":
options[name]["elements"] = help["elements"]
if help["type"] == "list": # type: ignore[reportOperatorIssue]
options[name]["elements"] = help["elements"] # type: ignore[reportIndexIssue]
if not options[name]["required"]:
options[name]["default"] = []
if "default" in help:
options[name]["default"] = help["default"]
if "options" in help:
options[name]["options"] = docify(help["options"])
if "choices" in help:
options[name]["choices"] = tuple(help["choices"])
if "default" in help: # type: ignore[reportOperatorIssue]
options[name]["default"] = help["default"] # type: ignore[reportIndexIssue]
if "options" in help and help["options"] != {}: # type: ignore[reportOperatorIssue]
options[name]["options"] = docify(help["options"]) # type: ignore[reportIndexIssue]
if "choices" in help and len(help["choices"]) > 0: # type: ignore[reportOperatorIssue]
options[name]["choices"] = tuple(help["choices"]) # type: ignore[reportIndexIssue]
return options
@ -54,6 +62,10 @@ class AnsibleModule(object):
result: dict
#: the specification of the arguments. Subclasses that are usable Modules must set this value.
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`
_common_args: ClassVar[dict[str, Any]] = dict()
@ -62,7 +74,11 @@ class AnsibleModule(object):
"""params is an wrapper for the module.params"""
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)
specs = dict()
specs.update(deepcopy(self._common_args))
@ -71,15 +87,20 @@ class AnsibleModule(object):
specs["argument_spec"].update(modspec["argument_spec"])
del modspec["argument_spec"]
specs.update(modspec)
self.module = basic.AnsibleModule(**specs)
self.tmpdir = pathlib.Path(self.module.tmpdir)
self.modspec = specs.copy()
specs.pop("deprecated", None)
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):
"""sets an value for the result"""
self.result[key] = value
@overload
def diff(self, diff: Dict[str, str]): # pragma: nocover
def diff(self, diff: TypedDiff): # pragma: nocover
pass
@overload
@ -107,7 +128,7 @@ class AnsibleModule(object):
if diff is not None and not any((before is not None, after is not None)):
pass
elif all((before is not None, after is not None, diff is None)):
diff = dict(
diff = TypedDiff(
before=before,
after=after,
)
@ -158,12 +179,8 @@ class AnsibleModule(object):
except Exception as exc:
import traceback
self.module.fail_json(
"".join(traceback.format_exception(type(exc), exc, exc.__traceback__)),
**self.result,
)
self.module.exit_json(**self.result)
raise Exception("exit_json failed")
self.fail("".join(traceback.format_exception(type(exc), exc, exc.__traceback__)))
self.exit()
@classmethod
def doc(cls) -> str:
@ -175,26 +192,81 @@ class AnsibleModule(object):
doc = cls.__doc__
if doc is None:
doc = ""
specs = dict()
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)
mod = cls(documentation=True)
options = docify(mod.modspec["argument_spec"])
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(
yaml.safe_dump(
dict(
module=cls.name,
short_description=docu[0],
description=docu,
options=options,
),
documentation,
stream=None,
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):
#: path of the unitfile managed by this module
@ -229,30 +301,37 @@ class SystemdUnitModule(AnsibleModule):
),
)
#: if defined it will be called after run has changed the unitfile
post: Optional[Callable[[], None]]
post: Optional[Callable[[], None]] = None
#: generates the install section if the unit is installable
install: ClassVar[Optional[Callable[["SystemdUnitModule"], str]]]
install: ClassVar[Optional[Callable[["SystemdUnitModule"], str | None]]] = None
def unit(self) -> str: # pragma: nocover
raise NotImplementedError()
def header(self) -> str:
header = "[Unit]\n"
header += "".join(
self.map_param(
description="Description",
documentation="Documentation",
requires="Requires",
wants="Wants",
partof="PartOf",
before="Before",
after="After",
)
def header(self) -> Optional[str]:
parts = self.map_param(
description="Description",
documentation="Documentation",
requires="Requires",
wants="Wants",
partof="PartOf",
before="Before",
after="After",
)
if len(parts) == 0:
return None
header = "[Unit]\n" + "".join(parts)
return header
def map_param(self, **parammap: str):
def _unit(self, *parts: Optional[str]) -> 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"""
output: list[str] = []
for param, key in parammap.items():
@ -264,38 +343,10 @@ class SystemdUnitModule(AnsibleModule):
output.append(f"{key}={systemdbool(params)}\n")
return output
def unitfile_gen(self): # pragma: nocover
def unitfile_gen(self) -> None: # pragma: nocover
path = self.tmpdir / "newunit"
with open(path, "w") as 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
self.set("unitfile", self.unitfile.as_posix())
@ -316,10 +367,7 @@ class SystemdUnitModule(AnsibleModule):
self.check()
if not self.changed:
return
self.module.atomic_move(
src=(self.tmpdir / "newunit").as_posix(),
dest=self.unitfile.as_posix(),
)
self.move_file(self.tmpdir / "newunit", self.unitfile)
if hasattr(self, "post") and self.post is not None:
self.post()
@ -343,13 +391,15 @@ def installable(_class: Type[SystemdUnitModule]): # pragma: nocover
)
specs["argument_spec"].update(arguments)
def install(self: SystemdUnitModule) -> str:
output = "[Install]\n"
def install(self: SystemdUnitModule) -> Optional[str]:
parts = []
for argument, key in _INSTALL_MAPPING.items():
if self.get(argument, False):
for unit in self.get(argument): # type: ignore
output += "{}={}\n".format(key, unit)
return output
parts.append("{}={}\n".format(key, unit))
if len(parts) == 0:
return None
return "[Install]\n" + "".join(parts)
_class.install = install
_class.module_spec = specs

Datei anzeigen

@ -7,23 +7,20 @@ name = "ansible-module"
dynamic = ["version"]
description = 'Helps with developing modules for ansible in an easier manner'
readme = "README.md"
requires-python = ">=3.7"
requires-python = ">=3.11"
license = "MIT"
keywords = []
authors = [{ name = "Sebastian Tobie", email = "sebastian@sebastian-tobie.de" }]
classifiers = [
"Development Status :: 4 - Beta",
"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 :: Implementation :: CPython",
"Programming Language :: Python :: Implementation :: PyPy",
]
dependencies = []
dependencies = ["ansible>=11.3.0"]
[tool.hatch.build]
directory = "dist/python"
[project.urls]
Documentation = "https://github.com/unknown/ansible-module#readme"
Issues = "https://github.com/unknown/ansible-module/issues"
@ -146,10 +143,11 @@ exclude_lines = [
]
[tool.mypy]
disable_error_code = "import-untyped"
python_version = "3.11"
warn_return_any = true
warn_unused_configs = true
[[tool.mypy.overrides]]
module = ["ansible", "ansible.module_utils", "ansible.module_utils.basic"]
ignore_missing_imports = true

Datei anzeigen

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

Datei anzeigen

@ -4,13 +4,18 @@ import pathlib
import re
import sys
from .module_utils.module import AnsibleModule
sys.path.append(".")
mindocstring = "DOCUMENTATION = ''''''"
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():
def main() -> None:
try:
modules = list(moduledir.iterdir())
except:
@ -20,6 +25,7 @@ def main():
if modfile.name in ("__init__.py", "__pycache__", "unit.py.example"):
continue
mod = importlib.import_module(".".join((modfile.parts[:-1]) + (modfile.stem,)))
module: AnsibleModule
if hasattr(mod, "Module"):
module = mod.Module
elif hasattr(mod, "__module_name__"):
@ -29,6 +35,7 @@ def main():
continue
try:
moddoc = module.doc()
returns = module.returns()
except AttributeError:
print("Broken module. skipping {}".format(modfile))
continue
@ -36,16 +43,37 @@ def main():
print("Error in documentation of module {}: {}".format(modfile, e))
continue
moddata = modfile.read_text()
match = regex.search(moddata)
if not match:
changed = False
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(
"no Documentation set for module {}. Please add at least \"{}\" to the file".format(
modfile.stem, mindocstring
)
)
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(newmod)
modfile.write_text(moddata)
print("updated the documentation of module {}".format(module.name))

Datei anzeigen

@ -3,11 +3,5 @@ user=$(yq -r .namespace galaxy.yml)
package=$(yq -r .name galaxy.yml)
version=$(yq -r .version galaxy.yml)
printf "Namespace: %s\nPackage: %s\nVersion: %s\n" $user $package $version
upload(){
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/* ; do
upload "$file"
done
ansible-galaxy collection publish dist/galaxy/*
hatch publish -r ansible dist/python/*