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 fragpull/8971/head
parent
96dfb89b01
commit
c4e2b73193
|
@ -310,6 +310,8 @@ files:
|
||||||
labels: module_utils
|
labels: module_utils
|
||||||
$module_utils/btrfs.py:
|
$module_utils/btrfs.py:
|
||||||
maintainers: gnfzdz
|
maintainers: gnfzdz
|
||||||
|
$module_utils/cmd_runner_fmt.py:
|
||||||
|
maintainers: russoz
|
||||||
$module_utils/cmd_runner.py:
|
$module_utils/cmd_runner.py:
|
||||||
maintainers: russoz
|
maintainers: russoz
|
||||||
$module_utils/deps.py:
|
$module_utils/deps.py:
|
||||||
|
|
|
@ -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).
|
|
@ -7,10 +7,10 @@ from __future__ import absolute_import, division, print_function
|
||||||
__metaclass__ = type
|
__metaclass__ = type
|
||||||
|
|
||||||
import os
|
import os
|
||||||
from functools import wraps
|
|
||||||
|
|
||||||
from ansible.module_utils.common.collections import is_sequence
|
from ansible.module_utils.common.collections import is_sequence
|
||||||
from ansible.module_utils.common.locale import get_best_parsable_locale
|
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):
|
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 "<ArgFormat: func={0}, ignore_none={1}, ignore_missing_value={2}>".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):
|
class CmdRunner(object):
|
||||||
"""
|
"""
|
||||||
Wrapper for ``AnsibleModule.run_command()``.
|
Wrapper for ``AnsibleModule.run_command()``.
|
||||||
|
@ -215,8 +109,8 @@ class CmdRunner(object):
|
||||||
arg_formats = {}
|
arg_formats = {}
|
||||||
self.arg_formats = {}
|
self.arg_formats = {}
|
||||||
for fmt_name, fmt in arg_formats.items():
|
for fmt_name, fmt in arg_formats.items():
|
||||||
if not isinstance(fmt, _ArgFormat):
|
if not cmd_runner_fmt.is_argformat(fmt):
|
||||||
fmt = _Format.as_func(func=fmt, ignore_none=True)
|
fmt = cmd_runner_fmt.as_func(func=fmt, ignore_none=True)
|
||||||
self.arg_formats[fmt_name] = fmt
|
self.arg_formats[fmt_name] = fmt
|
||||||
self.check_rc = check_rc
|
self.check_rc = check_rc
|
||||||
if force_lang == "auto":
|
if force_lang == "auto":
|
||||||
|
@ -350,6 +244,3 @@ class _CmdRunnerContext(object):
|
||||||
|
|
||||||
def __exit__(self, exc_type, exc_val, exc_tb):
|
def __exit__(self, exc_type, exc_val, exc_tb):
|
||||||
return False
|
return False
|
||||||
|
|
||||||
|
|
||||||
cmd_runner_fmt = _Format()
|
|
||||||
|
|
|
@ -0,0 +1,123 @@
|
||||||
|
# -*- coding: utf-8 -*-
|
||||||
|
# Copyright (c) 2024, Alexei Znamensky <russoz@gmail.com>
|
||||||
|
# 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 "<ArgFormat: func={0}, ignore_none={1}, ignore_missing_value={2}>".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)
|
Loading…
Reference in New Issue