cmd_runner_fmt: refactor out to its own file

.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:

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).

plugins/module_utils/cmd_runner.py

@@ -7,10 +7,10 @@
 __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 @@ def __str__(self):
         )
 
 
-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):
     """
     Wrapper for ``AnsibleModule.run_command()``.
@@ -215,8 +109,8 @@ def __init__(self, module, command, arg_formats=None, default_args_order=(),
             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 @@ def __enter__(self):
 
     def __exit__(self, exc_type, exc_val, exc_tb):
         return False
-
-
-cmd_runner_fmt = _Format()

plugins/module_utils/cmd_runner_fmt.py

@@ -0,0 +1,123 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2024, Alexei Znamensky <[email protected]>
+# 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)