From c4e2b731939e56a24870e9e3455381dc33ea3040 Mon Sep 17 00:00:00 2001 From: Alexei Znamensky <103110+russoz@users.noreply.github.com> Date: Thu, 3 Oct 2024 07:24:24 +1300 Subject: [PATCH] cmd_runner_fmt: refactor out to its own file (#8964) * cmd_runner_fmt: refactor out to its own file * add new file to BOTMETA.yml * add changelog frag --- .github/BOTMETA.yml | 2 + .../8964-cmd-runner-argformat-refactor.yml | 2 + plugins/module_utils/cmd_runner.py | 115 +--------------- plugins/module_utils/cmd_runner_fmt.py | 123 ++++++++++++++++++ 4 files changed, 130 insertions(+), 112 deletions(-) create mode 100644 changelogs/fragments/8964-cmd-runner-argformat-refactor.yml create mode 100644 plugins/module_utils/cmd_runner_fmt.py diff --git a/.github/BOTMETA.yml b/.github/BOTMETA.yml index c9326fa75a..10f2aee95b 100644 --- a/.github/BOTMETA.yml +++ b/.github/BOTMETA.yml @@ -310,6 +310,8 @@ files: labels: module_utils $module_utils/btrfs.py: maintainers: gnfzdz + $module_utils/cmd_runner_fmt.py: + maintainers: russoz $module_utils/cmd_runner.py: maintainers: russoz $module_utils/deps.py: diff --git a/changelogs/fragments/8964-cmd-runner-argformat-refactor.yml b/changelogs/fragments/8964-cmd-runner-argformat-refactor.yml new file mode 100644 index 0000000000..be8adf25e3 --- /dev/null +++ b/changelogs/fragments/8964-cmd-runner-argformat-refactor.yml @@ -0,0 +1,2 @@ +minor_changes: + - cmd_runner module utils - refactor argument formatting code to its own Python module (https://github.com/ansible-collections/community.general/pull/8964). diff --git a/plugins/module_utils/cmd_runner.py b/plugins/module_utils/cmd_runner.py index 5cd4f6b957..10278964bb 100644 --- a/plugins/module_utils/cmd_runner.py +++ b/plugins/module_utils/cmd_runner.py @@ -7,10 +7,10 @@ from __future__ import absolute_import, division, print_function __metaclass__ = type import os -from functools import wraps from ansible.module_utils.common.collections import is_sequence from ansible.module_utils.common.locale import get_best_parsable_locale +from ansible_collections.community.general.plugins.module_utils import cmd_runner_fmt def _ensure_list(value): @@ -88,112 +88,6 @@ class FormatError(CmdRunnerException): ) -class _ArgFormat(object): - # DEPRECATION: set default value for ignore_none to True in community.general 12.0.0 - def __init__(self, func, ignore_none=None, ignore_missing_value=False): - self.func = func - self.ignore_none = ignore_none - self.ignore_missing_value = ignore_missing_value - - # DEPRECATION: remove parameter ctx_ignore_none in community.general 12.0.0 - def __call__(self, value, ctx_ignore_none=True): - # DEPRECATION: replace ctx_ignore_none with True in community.general 12.0.0 - ignore_none = self.ignore_none if self.ignore_none is not None else ctx_ignore_none - if value is None and ignore_none: - return [] - f = self.func - return [str(x) for x in f(value)] - - def __str__(self): - return "".format( - self.func, - self.ignore_none, - self.ignore_missing_value, - ) - - def __repr__(self): - return str(self) - - -class _Format(object): - @staticmethod - def as_bool(args_true, args_false=None, ignore_none=None): - if args_false is not None: - if ignore_none is None: - ignore_none = False - else: - args_false = [] - return _ArgFormat(lambda value: _ensure_list(args_true) if value else _ensure_list(args_false), ignore_none=ignore_none) - - @staticmethod - def as_bool_not(args): - return _Format.as_bool([], args, ignore_none=False) - - @staticmethod - def as_optval(arg, ignore_none=None): - return _ArgFormat(lambda value: ["{0}{1}".format(arg, value)], ignore_none=ignore_none) - - @staticmethod - def as_opt_val(arg, ignore_none=None): - return _ArgFormat(lambda value: [arg, value], ignore_none=ignore_none) - - @staticmethod - def as_opt_eq_val(arg, ignore_none=None): - return _ArgFormat(lambda value: ["{0}={1}".format(arg, value)], ignore_none=ignore_none) - - @staticmethod - def as_list(ignore_none=None, min_len=0, max_len=None): - def func(value): - value = _ensure_list(value) - if len(value) < min_len: - raise ValueError("Parameter must have at least {0} element(s)".format(min_len)) - if max_len is not None and len(value) > max_len: - raise ValueError("Parameter must have at most {0} element(s)".format(max_len)) - return value - return _ArgFormat(func, ignore_none=ignore_none) - - @staticmethod - def as_fixed(args): - return _ArgFormat(lambda value: _ensure_list(args), ignore_none=False, ignore_missing_value=True) - - @staticmethod - def as_func(func, ignore_none=None): - return _ArgFormat(func, ignore_none=ignore_none) - - @staticmethod - def as_map(_map, default=None, ignore_none=None): - if default is None: - default = [] - return _ArgFormat(lambda value: _ensure_list(_map.get(value, default)), ignore_none=ignore_none) - - @staticmethod - def unpack_args(func): - @wraps(func) - def wrapper(v): - return func(*v) - return wrapper - - @staticmethod - def unpack_kwargs(func): - @wraps(func) - def wrapper(v): - return func(**v) - return wrapper - - @staticmethod - def stack(fmt): - @wraps(fmt) - def wrapper(*args, **kwargs): - new_func = fmt(ignore_none=True, *args, **kwargs) - - def stacking(value): - stack = [new_func(v) for v in value if v] - stack = [x for args in stack for x in args] - return stack - return _ArgFormat(stacking, ignore_none=True) - return wrapper - - class CmdRunner(object): """ Wrapper for ``AnsibleModule.run_command()``. @@ -215,8 +109,8 @@ class CmdRunner(object): arg_formats = {} self.arg_formats = {} for fmt_name, fmt in arg_formats.items(): - if not isinstance(fmt, _ArgFormat): - fmt = _Format.as_func(func=fmt, ignore_none=True) + if not cmd_runner_fmt.is_argformat(fmt): + fmt = cmd_runner_fmt.as_func(func=fmt, ignore_none=True) self.arg_formats[fmt_name] = fmt self.check_rc = check_rc if force_lang == "auto": @@ -350,6 +244,3 @@ class _CmdRunnerContext(object): def __exit__(self, exc_type, exc_val, exc_tb): return False - - -cmd_runner_fmt = _Format() diff --git a/plugins/module_utils/cmd_runner_fmt.py b/plugins/module_utils/cmd_runner_fmt.py new file mode 100644 index 0000000000..bd6d00a15d --- /dev/null +++ b/plugins/module_utils/cmd_runner_fmt.py @@ -0,0 +1,123 @@ +# -*- coding: utf-8 -*- +# Copyright (c) 2024, Alexei Znamensky +# GNU General Public License v3.0+ (see LICENSES/GPL-3.0-or-later.txt or https://www.gnu.org/licenses/gpl-3.0.txt) +# SPDX-License-Identifier: GPL-3.0-or-later + +from __future__ import absolute_import, division, print_function +__metaclass__ = type + +from functools import wraps + +from ansible.module_utils.common.collections import is_sequence + + +def _ensure_list(value): + return list(value) if is_sequence(value) else [value] + + +class _ArgFormat(object): + # DEPRECATION: set default value for ignore_none to True in community.general 12.0.0 + def __init__(self, func, ignore_none=None, ignore_missing_value=False): + self.func = func + self.ignore_none = ignore_none + self.ignore_missing_value = ignore_missing_value + + # DEPRECATION: remove parameter ctx_ignore_none in community.general 12.0.0 + def __call__(self, value, ctx_ignore_none=True): + # DEPRECATION: replace ctx_ignore_none with True in community.general 12.0.0 + ignore_none = self.ignore_none if self.ignore_none is not None else ctx_ignore_none + if value is None and ignore_none: + return [] + f = self.func + return [str(x) for x in f(value)] + + def __str__(self): + return "".format( + self.func, + self.ignore_none, + self.ignore_missing_value, + ) + + def __repr__(self): + return str(self) + + +def as_bool(args_true, args_false=None, ignore_none=None): + if args_false is not None: + if ignore_none is None: + ignore_none = False + else: + args_false = [] + return _ArgFormat(lambda value: _ensure_list(args_true) if value else _ensure_list(args_false), ignore_none=ignore_none) + + +def as_bool_not(args): + return as_bool([], args, ignore_none=False) + + +def as_optval(arg, ignore_none=None): + return _ArgFormat(lambda value: ["{0}{1}".format(arg, value)], ignore_none=ignore_none) + + +def as_opt_val(arg, ignore_none=None): + return _ArgFormat(lambda value: [arg, value], ignore_none=ignore_none) + + +def as_opt_eq_val(arg, ignore_none=None): + return _ArgFormat(lambda value: ["{0}={1}".format(arg, value)], ignore_none=ignore_none) + + +def as_list(ignore_none=None, min_len=0, max_len=None): + def func(value): + value = _ensure_list(value) + if len(value) < min_len: + raise ValueError("Parameter must have at least {0} element(s)".format(min_len)) + if max_len is not None and len(value) > max_len: + raise ValueError("Parameter must have at most {0} element(s)".format(max_len)) + return value + return _ArgFormat(func, ignore_none=ignore_none) + + +def as_fixed(args): + return _ArgFormat(lambda value: _ensure_list(args), ignore_none=False, ignore_missing_value=True) + + +def as_func(func, ignore_none=None): + return _ArgFormat(func, ignore_none=ignore_none) + + +def as_map(_map, default=None, ignore_none=None): + if default is None: + default = [] + return _ArgFormat(lambda value: _ensure_list(_map.get(value, default)), ignore_none=ignore_none) + + +def unpack_args(func): + @wraps(func) + def wrapper(v): + return func(*v) + return wrapper + + +def unpack_kwargs(func): + @wraps(func) + def wrapper(v): + return func(**v) + return wrapper + + +def stack(fmt): + @wraps(fmt) + def wrapper(*args, **kwargs): + new_func = fmt(ignore_none=True, *args, **kwargs) + + def stacking(value): + stack = [new_func(v) for v in value if v] + stack = [x for args in stack for x in args] + return stack + return _ArgFormat(stacking, ignore_none=True) + return wrapper + + +def is_argformat(fmt): + return isinstance(fmt, _ArgFormat)