pipx: use CmdRunner

.github/BOTMETA.yml

@@ -290,6 +290,9 @@ files:
   $module_utils/oracle/oci_utils.py:
     maintainers: $team_oracle
     labels: cloud
+  $module_utils/pipx.py:
+    maintainers: russoz
+    labels: pipx
   $module_utils/pure.py:
     maintainers: $team_purestorage
     labels: pure pure_storage

changelogs/fragments/5085-pipx-use-cmd-runner.yaml

@@ -0,0 +1,3 @@
+minor_changes:
+  - pipx module utils - created new module util ``pipx`` providing a ``cmd_runner`` specific for the ``pipx`` module (https://github.com/ansible-collections/community.general/pull/5085).
+  - pipx - changed implementation to use ``cmd_runner`` (https://github.com/ansible-collections/community.general/pull/5085).

plugins/module_utils/pipx.py

@@ -0,0 +1,50 @@
+# -*- coding: utf-8 -*-
+# Copyright (c) 2022, 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 ansible.module_utils.parsing.convert_bool import boolean
+from ansible_collections.community.general.plugins.module_utils.cmd_runner import CmdRunner, cmd_runner_fmt as fmt
+
+
+_state_map = dict(
+    install='install',
+    present='install',
+    uninstall='uninstall',
+    absent='uninstall',
+    uninstall_all='uninstall-all',
+    inject='inject',
+    upgrade='upgrade',
+    upgrade_all='upgrade-all',
+    reinstall='reinstall',
+    reinstall_all='reinstall-all',
+)
+
+
+def pipx_runner(module, command, **kwargs):
+    runner = CmdRunner(
+        module,
+        command=command,
+        arg_formats=dict(
+
+            state=fmt.as_map(_state_map),
+            name=fmt.as_list(),
+            name_source=fmt.as_func(fmt.unpack_args(lambda n, s: [s] if s else [n])),
+            install_deps=fmt.as_bool("--include-deps"),
+            inject_packages=fmt.as_list(),
+            force=fmt.as_bool("--force"),
+            include_injected=fmt.as_bool("--include-injected"),
+            index_url=fmt.as_opt_val('--index-url'),
+            python=fmt.as_opt_val('--python'),
+            _list=fmt.as_fixed(['list', '--include-injected', '--json']),
+            editable=fmt.as_bool("--editable"),
+            pip_args=fmt.as_opt_val('--pip-args'),
+        ),
+        environ_update={'USE_EMOJI': '0'},
+        check_rc=True,
+        **kwargs
+    )
+    return runner

plugins/modules/packaging/language/pipx.py

@@ -133,22 +133,13 @@
 
 import json
 
-from ansible_collections.community.general.plugins.module_utils.module_helper import (
-    CmdStateModuleHelper, ArgFormat
-)
-from ansible.module_utils.facts.compat import ansible_facts
-
+from ansible_collections.community.general.plugins.module_utils.module_helper import StateModuleHelper
+from ansible_collections.community.general.plugins.module_utils.pipx import pipx_runner
 
-_state_map = dict(
-    present='install',
-    absent='uninstall',
-    uninstall_all='uninstall-all',
-    upgrade_all='upgrade-all',
-    reinstall_all='reinstall-all',
-)
+from ansible.module_utils.facts.compat import ansible_facts
 
 
-class PipX(CmdStateModuleHelper):
+class PipX(StateModuleHelper):
     output_params = ['name', 'source', 'index_url', 'force', 'installdeps']
     module = dict(
         argument_spec=dict(
@@ -172,27 +163,11 @@ class PipX(CmdStateModuleHelper):
             ('state', 'install', ['name']),
             ('state', 'absent', ['name']),
             ('state', 'uninstall', ['name']),
+            # missing upgrade and reinstall requiring 'name'
             ('state', 'inject', ['name', 'inject_packages']),
         ],
         supports_check_mode=True,
     )
-    command_args_formats = dict(
-        state=dict(fmt=lambda v: [_state_map.get(v, v)]),
-        name_source=dict(fmt=lambda n, s: [s] if s else [n], stars=1),
-        install_deps=dict(fmt="--include-deps", style=ArgFormat.BOOLEAN),
-        inject_packages=dict(fmt=lambda v: v),
-        force=dict(fmt="--force", style=ArgFormat.BOOLEAN),
-        include_injected=dict(fmt="--include-injected", style=ArgFormat.BOOLEAN),
-        index_url=dict(fmt=('--index-url', '{0}'),),
-        python=dict(fmt=('--python', '{0}'),),
-        _list=dict(fmt=('list', '--include-injected', '--json'), style=ArgFormat.BOOLEAN),
-        editable=dict(fmt="--editable", style=ArgFormat.BOOLEAN),
-        pip_args=dict(fmt=('--pip-args', '{0}'),),
-    )
-    check_rc = True
-    run_command_fixed_options = dict(
-        environ_update={'USE_EMOJI': '0'}
-    )
 
     def _retrieve_installed(self):
         def process_list(rc, out, err):
@@ -210,8 +185,7 @@ def process_list(rc, out, err):
                 }
             return results
 
-        installed = self.run_command(params=[{'_list': True}], process_output=process_list,
-                                     publish_rc=False, publish_out=False, publish_err=False, publish_cmd=False)
+        installed = self.runner('_list', output_process=process_list).run(_list=1)
 
         if self.vars.name is not None:
             app_list = installed.get(self.vars.name)
@@ -228,19 +202,26 @@ def __init_module__(self):
         else:
             facts = ansible_facts(self.module, gather_subset=['python'])
             self.command = [facts['python']['executable'], '-m', 'pipx']
+        self.runner = pipx_runner(self.module, self.command)
 
         self.vars.set('application', self._retrieve_installed(), change=True, diff=True)
 
     def __quit_module__(self):
         self.vars.application = self._retrieve_installed()
 
+    def _capture_results(self, ctx):
+        self.vars.stdout = ctx.results_out
+        self.vars.stderr = ctx.results_err
+        self.vars.cmd = ctx.cmd
+        if self.verbosity >= 4:
+            self.vars.run_info = ctx.run_info
+
     def state_install(self):
         if not self.vars.application or self.vars.force:
             self.changed = True
-            if not self.module.check_mode:
-                self.run_command(params=[
-                    'state', 'index_url', 'install_deps', 'force', 'python', 'editable', 'pip_args',
-                    {'name_source': [self.vars.name, self.vars.source]}])
+            with self.runner('state index_url install_deps force python editable pip_args name_source', check_mode_skip=True) as ctx:
+                ctx.run(name_source=[self.vars.name, self.vars.source])
+                self._capture_results(ctx)
 
     state_present = state_install
 
@@ -249,43 +230,52 @@ def state_upgrade(self):
             self.do_raise("Trying to upgrade a non-existent application: {0}".format(self.vars.name))
         if self.vars.force:
             self.changed = True
-        if not self.module.check_mode:
-            self.run_command(params=['state', 'index_url', 'install_deps', 'force', 'editable', 'pip_args', 'name'])
+
+        with self.runner('state index_url install_deps force editable pip_args name', check_mode_skip=True) as ctx:
+            ctx.run()
+            self._capture_results(ctx)
 
     def state_uninstall(self):
-        if self.vars.application and not self.module.check_mode:
-            self.run_command(params=['state', 'name'])
+        if self.vars.application:
+            with self.runner('state name', check_mode_skip=True) as ctx:
+                ctx.run()
+                self._capture_results(ctx)
 
     state_absent = state_uninstall
 
     def state_reinstall(self):
         if not self.vars.application:
             self.do_raise("Trying to reinstall a non-existent application: {0}".format(self.vars.name))
         self.changed = True
-        if not self.module.check_mode:
-            self.run_command(params=['state', 'name', 'python'])
+        with self.runner('state name python', check_mode_skip=True) as ctx:
+            ctx.run()
+            self._capture_results(ctx)
 
     def state_inject(self):
         if not self.vars.application:
             self.do_raise("Trying to inject packages into a non-existent application: {0}".format(self.vars.name))
         if self.vars.force:
             self.changed = True
-        if not self.module.check_mode:
-            self.run_command(params=['state', 'index_url', 'force', 'editable', 'pip_args', 'name', 'inject_packages'])
+        with self.runner('state index_url force editable pip_args name inject_packages', check_mode_skip=True) as ctx:
+            ctx.run()
+            self._capture_results(ctx)
 
     def state_uninstall_all(self):
-        if not self.module.check_mode:
-            self.run_command(params=['state'])
+        with self.runner('state', check_mode_skip=True) as ctx:
+            ctx.run()
+            self._capture_results(ctx)
 
     def state_reinstall_all(self):
-        if not self.module.check_mode:
-            self.run_command(params=['state', 'python'])
+        with self.runner('state python', check_mode_skip=True) as ctx:
+            ctx.run()
+            self._capture_results(ctx)
 
     def state_upgrade_all(self):
         if self.vars.force:
             self.changed = True
-        if not self.module.check_mode:
-            self.run_command(params=['state', 'include_injected', 'force'])
+        with self.runner('state include_injected force', check_mode_skip=True) as ctx:
+            ctx.run()
+            self._capture_results(ctx)
 
 
 def main():