[ContainerApp] az containerapp replica count: app/revision scope

src/containerapp/HISTORY.rst

@@ -7,6 +7,7 @@ upcoming
 * 'az containerapp job create': Fix AttributeError when --trigger-type is None
 * 'az containerapp update': fix bug for mounting secret volumes using --secret-volume-mount
 * 'az containerapp compose create': fixed an issue where the environment's resource group was not resolved from --environment when the input value was a resource id.
+* 'az containerapp replica count', returns the replica count of a container app
 
 0.3.41
 ++++++

src/containerapp/azext_containerapp/_help.py

@@ -261,6 +261,18 @@
           az containerapp replica show -n MyContainerapp -g MyResourceGroup --replica MyReplica --revision MyRevision
 """
 
+helps['containerapp replica count'] = """
+    type: command
+    short-summary: Count of a container app's replica(s)
+    examples:
+    - name: Count replicas of a particular revision
+      text: |
+          az containerapp replica count -n MyContainerapp -g MyResourceGroup --revision MyRevision
+    - name: Count replicas of the latest revision
+      text: |
+          az containerapp replica count -n MyContainerapp -g MyResourceGroup
+"""
+
 # Revision Commands
 helps['containerapp revision'] = """
     type: group

src/containerapp/azext_containerapp/commands.py

@@ -32,6 +32,7 @@ def load_command_table(self, _):
     with self.command_group('containerapp replica') as g:
         g.custom_show_command('show', 'get_replica')  # TODO implement the table transformer
         g.custom_command('list', 'list_replicas')
+        g.custom_command('count', 'count_replicas', is_preview=True)
 
     with self.command_group('containerapp logs') as g:
         g.custom_show_command('show', 'stream_containerapp_logs', validator=validate_ssh)

src/containerapp/azext_containerapp/custom.py

@@ -3347,24 +3347,53 @@ def remove_dapr_component(cmd, resource_group_name, dapr_component_name, environ
 
 
 def list_replicas(cmd, resource_group_name, name, revision=None):
-    app = ContainerAppClient.show(cmd, resource_group_name, name)
-    if not revision:
-        revision = app["properties"]["latestRevisionName"]
-    return ContainerAppClient.list_replicas(cmd=cmd,
-                                            resource_group_name=resource_group_name,
-                                            container_app_name=name,
-                                            revision_name=revision)
+
+    try:
+        app = ContainerAppClient.show(cmd, resource_group_name, name)
+        if not revision:
+            revision = app["properties"]["latestRevisionName"]
+        return ContainerAppClient.list_replicas(cmd=cmd,
+                                                resource_group_name=resource_group_name,
+                                                container_app_name=name,
+                                                revision_name=revision)
+    except Exception as e:
+        handle_raw_exception(e)
+
+
+def count_replicas(cmd, resource_group_name, name, revision=None):
+
+    try:
+        app = ContainerAppClient.show(cmd, resource_group_name, name)
+        if not revision:
+            revision = safe_get(app, "properties", "latestRevisionName")
+            if not revision:
+                raise ValidationError("No revision found for containerapp.")
+    except Exception as e:
+        handle_raw_exception(e)
+
+    try:
+        count = len(ContainerAppClient.list_replicas(cmd=cmd,
+                                                     resource_group_name=resource_group_name,
+                                                     container_app_name=name,
+                                                     revision_name=revision))
+        return count
+    except Exception as e:
+        handle_raw_exception(e)
 
 
 def get_replica(cmd, resource_group_name, name, replica, revision=None):
-    app = ContainerAppClient.show(cmd, resource_group_name, name)
-    if not revision:
-        revision = app["properties"]["latestRevisionName"]
-    return ContainerAppClient.get_replica(cmd=cmd,
-                                          resource_group_name=resource_group_name,
-                                          container_app_name=name,
-                                          revision_name=revision,
-                                          replica_name=replica)
+
+    try:
+        app = ContainerAppClient.show(cmd, resource_group_name, name)
+        if not revision:
+            revision = app["properties"]["latestRevisionName"]
+        return ContainerAppClient.get_replica(cmd=cmd,
+                                              resource_group_name=resource_group_name,
+                                              container_app_name=name,
+                                              revision_name=revision,
+                                              replica_name=replica)
+    except Exception as e:
+        handle_raw_exception(e)
 
 
 def containerapp_ssh(cmd, resource_group_name, name, container=None, revision=None, replica=None, startup_command="sh"):