feat(cli): add --limit flag to cap number of apps per command

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: BradMclain

gitops/core.py

@@ -18,17 +18,19 @@
 
 
 @task
-def summary(ctx: Any, filter: str = "", exclude: str = "") -> None:
+def summary(ctx: Any, filter: str = "", exclude: str = "", limit: int = 0) -> None:
     """Produce a summary of apps, their tags, and their expected images & replicas.
     May not necessarily reflect actual app statuses if recent changes haven't yet been pushed to
     the remote, or the deployment has failed.
+    Provide `limit` to show only the first N matching apps.
     """
     get_apps(
         filter=filter,
         exclude=exclude,
         mode="PREVIEW",
         autoexclude_inactive=False,
         load_secrets=False,
+        limit=limit,
     )
 
 
@@ -45,6 +47,7 @@ def bump(  # noqa: C901
     redeploy: bool = False,
     skip_migrations: bool = False,
     skip_deploy: bool = False,
+    limit: int = 0,
 ) -> None:
     """Bump image tag on selected app(s).
     Provide `image_tag` to set to a specific image tag, or provide `prefix` to use latest image
@@ -54,6 +57,7 @@ def bump(  # noqa: C901
     Provide `redeploy` to redeploy servers even if nothing has changed.
     Provide `skip_migrations` to disable running migrations via helm hooks.
     Provide `skip_deploy` to skip deploying for non functional changes.
+    Provide `limit` to process only the first N matching apps.
     """
     prompt_message = "The following apps will have their image bumped"
     if image_tag:
@@ -68,6 +72,7 @@ def bump(  # noqa: C901
             message=f"{prompt_message}{colourise(':', Fore.LIGHTBLUE_EX)}",
             load_secrets=False,
             mode="PROMPT" if interactive else "SILENT",
+            limit=limit,
         )
     except AppOperationAborted:
         print(success_negative("Aborted."))
@@ -143,11 +148,13 @@ def redeploy(
     interactive: bool = True,
     push: bool = False,
     skip_migrations: bool = False,
+    limit: int = 0,
 ) -> None:
     """Force redeploy selected app(s) without changing their image tag.
     Updates a UUID bump field to trigger a new deployment.
     Provide `push` to automatically push the commit (and retry on conflict.)
     Provide `skip_migrations` to disable running migrations via helm hooks.
+    Provide `limit` to process only the first N matching apps.
     """
     try:
         apps = get_apps(
@@ -157,6 +164,7 @@ def redeploy(
             message=f"{colourise('The following apps will be redeployed:', Fore.LIGHTBLUE_EX)}",
             load_secrets=False,
             mode="PROMPT" if interactive else "SILENT",
+            limit=limit,
         )
     except AppOperationAborted:
         print(success_negative("Aborted."))
@@ -193,10 +201,12 @@ def command(
     interactive: bool = True,
     cpu: int = 0,
     memory: int = 0,
+    limit: int = 0,
 ) -> None:
     """Run command on selected app(s).
 
     eg. inv command customer,sandbox -e aesg "python manage.py migrate"
+    Provide `limit` to process only the first N matching apps.
     """
     try:
         apps = get_apps(
@@ -208,6 +218,7 @@ def command(
                 f" {colourise('will be run on the following apps:', Fore.LIGHTBLUE_EX)}"
             ),
             mode="PROMPT" if interactive else "SILENT",
+            limit=limit,
         )
     except AppOperationAborted:
         print(success_negative("Aborted."))
@@ -233,8 +244,10 @@ def command(
 
 
 @task
-def tag(ctx: Any, filter: str, tag: str, exclude: str = "") -> None:
-    """Set a tag on selected app(s)."""
+def tag(ctx: Any, filter: str, tag: str, exclude: str = "", limit: int = 0) -> None:
+    """Set a tag on selected app(s).
+    Provide `limit` to process only the first N matching apps.
+    """
     try:
         apps = get_apps(
             filter=filter,
@@ -245,6 +258,7 @@ def tag(ctx: Any, filter: str, tag: str, exclude: str = "") -> None:
                 f" {colourise('will be added to the following apps:', Fore.LIGHTBLUE_EX)}"
             ),
             load_secrets=False,
+            limit=limit,
         )
     except AppOperationAborted:
         print(success_negative("Aborted."))
@@ -259,8 +273,10 @@ def tag(ctx: Any, filter: str, tag: str, exclude: str = "") -> None:
 
 
 @task
-def untag(ctx: Any, filter: str, tag: str, exclude: str = "") -> None:
-    """Unset a tag from selected app(s)."""
+def untag(ctx: Any, filter: str, tag: str, exclude: str = "", limit: int = 0) -> None:
+    """Unset a tag from selected app(s).
+    Provide `limit` to process only the first N matching apps.
+    """
     try:
         apps = get_apps(
             filter=filter,
@@ -271,6 +287,7 @@ def untag(ctx: Any, filter: str, tag: str, exclude: str = "") -> None:
                 f" {colourise('will be removed from the following apps:', Fore.LIGHTBLUE_EX)}"
             ),
             load_secrets=False,
+            limit=limit,
         )
     except AppOperationAborted:
         print(success_negative("Aborted."))
@@ -323,12 +340,13 @@ def _sort_envs(envs: dict[str, Any]) -> dict[str, Any]:
 
 
 @task
-def setenv(ctx: Any, filter: str, values: str, exclude: str = "") -> None:
+def setenv(ctx: Any, filter: str, values: str, exclude: str = "", limit: int = 0) -> None:
     """Set one or more env vars on selected app(s).
 
     eg. inv setenv customer,sandbox BG_RUNNER=DRAMATIQ,BUMP=2
 
     NOTE: More broad-reaching environment changes should be made at the chart level.
+    Provide `limit` to process only the first N matching apps.
     """
     splitenvs = values.split(",")  # pardon the pun.
     formatted_splitenvs = "\n".join(splitenvs)
@@ -339,6 +357,7 @@ def setenv(ctx: Any, filter: str, values: str, exclude: str = "") -> None:
             message=(
                 f"{colourise('The env var(s)', Fore.LIGHTBLUE_EX)}\n{colourise(formatted_splitenvs, Fore.LIGHTYELLOW_EX)}\n{colourise('will be added to the following apps:', Fore.LIGHTBLUE_EX)}"
             ),
+            limit=limit,
         )
     except AppOperationAborted:
         print(success_negative("Aborted."))
@@ -366,12 +385,13 @@ def setenv(ctx: Any, filter: str, values: str, exclude: str = "") -> None:
 
 
 @task
-def unsetenv(ctx: Any, filter: str, values: str, exclude: str = "") -> None:
+def unsetenv(ctx: Any, filter: str, values: str, exclude: str = "", limit: int = 0) -> None:
     """Unset one or more env vars on selected app(s).
 
     eg. inv unsetenv customer,sandbox BG_RUNNER,BUMP
 
     NOTE: More broad-reaching environment changes should be made at the chart level.
+    Provide `limit` to process only the first N matching apps.
     """
     splitenvs = values.split(",")  # pardon the pun.
     formatted_splitenvs = "\n".join(splitenvs)
@@ -382,6 +402,7 @@ def unsetenv(ctx: Any, filter: str, values: str, exclude: str = "") -> None:
             message=(
                 f"{colourise('The env var(s)', Fore.LIGHTBLUE_EX)}\n{colourise(formatted_splitenvs, Fore.LIGHTYELLOW_EX)}\n{colourise('will be removed from the following apps:', Fore.LIGHTBLUE_EX)}"
             ),
+            limit=limit,
         )
     except AppOperationAborted:
         print(success_negative("Aborted."))
@@ -400,10 +421,11 @@ def unsetenv(ctx: Any, filter: str, values: str, exclude: str = "") -> None:
 
 
 @task
-def setcluster(ctx: Any, filter: str, cluster: str, exclude: str = "") -> None:
+def setcluster(ctx: Any, filter: str, cluster: str, exclude: str = "", limit: int = 0) -> None:
     """Move selected app(s) to given cluster.
 
     eg. inv setcluster customer,sandbox eks-prod
+    Provide `limit` to process only the first N matching apps.
     """
     try:
         apps = get_apps(
@@ -414,6 +436,7 @@ def setcluster(ctx: Any, filter: str, cluster: str, exclude: str = "") -> None:
                 f" {colourise(cluster, Fore.LIGHTYELLOW_EX)}"
                 f" {colourise('cluster:', Fore.LIGHTBLUE_EX)}"
             ),
+            limit=limit,
         )
     except AppOperationAborted:
         print(success_negative("Aborted."))

gitops/utils/apps.py

@@ -73,6 +73,7 @@ def get_apps(  # noqa: C901
     message: str | None = None,
     load_secrets: bool = True,
     encode_secrets: bool = True,
+    limit: int = 0,
 ) -> list[App]:
     """Return apps that contain ALL of the tags listed in `filter` and NONE of the tags listed in
     `exclude`. The incoming filter and exclude params may come in as a list or commastring.
@@ -133,6 +134,9 @@ def get_apps(  # noqa: C901
 
     validate_tags(filter | exclude, existing_tags)
 
+    if limit:  # 0 means no limit
+        apps = apps[:limit]
+
     if mode in ["PROMPT", "PREVIEW"]:
         if mode == "PROMPT" and message is None:
             message = "The following apps will be affected:"