Determine the compose version via a CLI call and not the docker API.

changelogs/fragments/1021-docker_compose_v2-version-detection.yml

@@ -0,0 +1,2 @@
+minor_changes:
+  - "docker_compose_v2* modules - determine compose version with ``docker compose version`` and only then fall back to ``docker info`` (https://github.com/ansible-collections/community.docker/pull/1021)."

plugins/module_utils/compose_v2.py

@@ -701,16 +701,7 @@ def __init__(self, client, min_version=MINIMUM_COMPOSE_VERSION):
         self.env_files = parameters['env_files']
         self.profiles = parameters['profiles']
 
-        compose = self.client.get_client_plugin_info('compose')
-        if compose is None:
-            self.fail('Docker CLI {0} does not have the compose plugin installed'.format(self.client.get_cli()))
-        if compose['Version'] == 'dev':
-            self.fail(
-                'Docker CLI {0} has a compose plugin installed, but it reports version "dev".'
-                ' Please use a version of the plugin that returns a proper version.'
-                .format(self.client.get_cli())
-            )
-        compose_version = compose['Version'].lstrip('v')
+        compose_version = self.get_compose_version()
         self.compose_version = LooseVersion(compose_version)
         if self.compose_version < LooseVersion(min_version):
             self.fail('Docker CLI {cli} has the compose plugin with version {version}; need version {min_version} or later'.format(
@@ -736,6 +727,33 @@ def __init__(self, client, min_version=MINIMUM_COMPOSE_VERSION):
         # more precisely in https://github.com/docker/compose/pull/11478
         self.use_json_events = self.compose_version >= LooseVersion('2.29.0')
 
+    def get_compose_version(self):
+        return self.get_compose_version_from_cli() or self.get_compose_version_from_api()
+
+    def get_compose_version_from_cli(self):
+        rc, version_info, stderr = self.client.call_cli('compose', 'version', '--format', 'json')
+        if rc:
+            return None
+        try:
+            version = json.loads(version_info)['version']
+            if version == 'dev':
+                return None
+            return version.lstrip('v')
+        except Exception:
+            return None
+
+    def get_compose_version_from_api(self):
+        compose = self.client.get_client_plugin_info('compose')
+        if compose is None:
+            self.fail('Docker CLI {0} does not have the compose plugin installed'.format(self.client.get_cli()))
+        if compose['Version'] == 'dev':
+            self.fail(
+                'Docker CLI {0} has a compose plugin installed, but it reports version "dev".'
+                ' Please use a version of the plugin that returns a proper version.'
+                .format(self.client.get_cli())
+            )
+        return compose['Version'].lstrip('v')
+
     def fail(self, msg, **kwargs):
         self.cleanup()
         self.client.fail(msg, **kwargs)