Add Pagination Support for Large ECS Deployments

pyproject.toml

@@ -1,6 +1,6 @@
 [project]
 name = "lazy-ecs"
-version = "0.1.19"
+version = "0.2.0"
 description = "A CLI tool for working with AWS services"
 readme = "README.md"
 authors = [

src/lazy_ecs/core/navigation.py

@@ -8,6 +8,8 @@
 from prompt_toolkit.keys import Keys
 from rich.console import Console
 
+PAGINATION_THRESHOLD = 30
+
 
 def parse_selection(selected: str | None) -> tuple[str, str, str]:
     """Parse selection into (type, value, extra). Returns ('unknown', selected, '') if no colon."""
@@ -120,3 +122,62 @@ def _(event: KeyPressEvent) -> None:
             question.application.key_bindings = merged_bindings
 
     return question.ask()
+
+
+def select_with_pagination(
+    prompt: str, choices: list[dict[str, str]], back_text: str | None, page_size: int = 25
+) -> str | None:
+    """Selection with pagination for large lists."""
+    total_items = len(choices)
+    total_pages = (total_items + page_size - 1) // page_size
+    current_page = 0
+
+    while True:
+        start_idx = current_page * page_size
+        end_idx = min(start_idx + page_size, total_items)
+        page_choices = choices[start_idx:end_idx]
+
+        page_prompt = f"{prompt} (Page {current_page + 1} of {total_pages})"
+
+        paginated_choices = []
+        for choice in page_choices:
+            paginated_choices.append(questionary.Choice(choice["name"], choice["value"]))
+
+        if current_page < total_pages - 1:
+            paginated_choices.append(
+                questionary.Choice(
+                    f"→ Next Page ({end_idx + 1}-{min(end_idx + page_size, total_items)})", "pagination:next"
+                )
+            )
+
+        if current_page > 0:
+            paginated_choices.append(
+                questionary.Choice(f"← Previous Page ({start_idx - page_size + 1}-{start_idx})", "pagination:previous")
+            )
+
+        if back_text:
+            paginated_choices.append(questionary.Choice(f"⬅️ {back_text}", "navigation:back"))
+
+        paginated_choices.append(questionary.Choice("❌ Exit", "navigation:exit"))
+
+        selected = questionary.select(
+            page_prompt, choices=paginated_choices, style=get_questionary_style(), use_shortcuts=False
+        ).ask()
+
+        if selected == "pagination:next":
+            current_page += 1
+        elif selected == "pagination:previous":
+            current_page -= 1
+        else:
+            return selected
+
+
+def select_with_auto_pagination(
+    prompt: str, choices: list[dict[str, str]], back_text: str | None, threshold: int = PAGINATION_THRESHOLD
+) -> str | None:
+    """Select with automatic pagination based on choice count.
+
+    Uses keyboard shortcuts for small lists (≤threshold), pagination for large lists (>threshold).
+    """
+    select_fn = select_with_pagination if len(choices) > threshold else select_with_navigation
+    return select_fn(prompt, choices, back_text)

src/lazy_ecs/core/utils.py

@@ -4,10 +4,14 @@
 
 from collections.abc import Iterator
 from contextlib import contextmanager
+from typing import TYPE_CHECKING, Literal
 
 from rich.console import Console
 from rich.spinner import Spinner
 
+if TYPE_CHECKING:
+    from mypy_boto3_ecs.client import ECSClient
+
 console = Console()
 
 
@@ -49,3 +53,29 @@ def show_spinner() -> Iterator[None]:
     spinner = Spinner("dots", style="cyan")
     with console.status(spinner):
         yield
+
+
+def paginate_aws_list(
+    client: ECSClient,
+    operation_name: Literal[
+        "list_account_settings",
+        "list_attributes",
+        "list_clusters",
+        "list_container_instances",
+        "list_services_by_namespace",
+        "list_services",
+        "list_task_definition_families",
+        "list_task_definitions",
+        "list_tasks",
+    ],
+    result_key: str,
+    **kwargs: str,
+) -> list[str]:
+    paginator = client.get_paginator(operation_name)  # type: ignore[no-matching-overload]
+    page_iterator = paginator.paginate(**kwargs)
+
+    results: list[str] = []
+    for page in page_iterator:
+        results.extend(page.get(result_key, []))
+
+    return results

src/lazy_ecs/features/cluster/cluster.py

@@ -5,7 +5,7 @@
 from typing import TYPE_CHECKING
 
 from ...core.base import BaseAWSService
-from ...core.utils import extract_name_from_arn
+from ...core.utils import extract_name_from_arn, paginate_aws_list
 
 if TYPE_CHECKING:
     from mypy_boto3_ecs.client import ECSClient
@@ -18,6 +18,5 @@ def __init__(self, ecs_client: ECSClient) -> None:
         super().__init__(ecs_client)
 
     def get_cluster_names(self) -> list[str]:
-        response = self.ecs_client.list_clusters()
-        cluster_arns = response.get("clusterArns", [])
+        cluster_arns = paginate_aws_list(self.ecs_client, "list_clusters", "clusterArns")
         return [extract_name_from_arn(arn) for arn in cluster_arns]

src/lazy_ecs/features/cluster/ui.py

@@ -5,7 +5,7 @@
 from rich.console import Console
 
 from ...core.base import BaseUIComponent
-from ...core.navigation import handle_navigation, select_with_navigation
+from ...core.navigation import handle_navigation, select_with_auto_pagination
 from ...core.utils import show_spinner
 from .cluster import ClusterService
 
@@ -27,18 +27,12 @@ def select_cluster(self) -> str:
             console.print("❌ No ECS clusters found", style="red")
             return ""
 
-        # Convert cluster names to choice format
         choices = [{"name": name, "value": name} for name in cluster_names]
 
-        selected = select_with_navigation(
-            "Select an ECS cluster:",
-            choices,
-            None,  # No back option for top-level menu
-        )
+        selected = select_with_auto_pagination("Select an ECS cluster:", choices, None)
 
-        # Handle navigation responses
         should_continue, _should_exit = handle_navigation(selected)
         if not should_continue:
-            return ""  # Exit was chosen
+            return ""
 
         return selected or ""

src/lazy_ecs/features/service/service.py

@@ -7,7 +7,7 @@
 
 from ...core.base import BaseAWSService
 from ...core.types import ServiceEvent, ServiceInfo
-from ...core.utils import determine_service_status, extract_name_from_arn
+from ...core.utils import determine_service_status, extract_name_from_arn, paginate_aws_list
 
 if TYPE_CHECKING:
     from typing import Any
@@ -23,8 +23,7 @@ def __init__(self, ecs_client: ECSClient) -> None:
         super().__init__(ecs_client)
 
     def get_services(self, cluster_name: str) -> list[str]:
-        response = self.ecs_client.list_services(cluster=cluster_name)
-        service_arns = response.get("serviceArns", [])
+        service_arns = paginate_aws_list(self.ecs_client, "list_services", "serviceArns", cluster=cluster_name)
         return [extract_name_from_arn(arn) for arn in service_arns]
 
     def get_service_info(self, cluster_name: str) -> list[ServiceInfo]:

src/lazy_ecs/features/service/ui.py

@@ -7,6 +7,7 @@
 from rich.table import Table
 
 from ...core.base import BaseUIComponent
+from ...core.navigation import select_with_auto_pagination
 from ...core.types import TaskInfo
 from ...core.utils import show_spinner
 from .actions import ServiceActions
@@ -34,7 +35,7 @@ def select_service(self, cluster_name: str) -> str | None:
 
         choices = [{"name": info["name"], "value": f"service:{info['name'].split(' ')[1]}"} for info in service_info]
 
-        return self.select_with_nav("Select a service:", choices, "Back to cluster selection")
+        return select_with_auto_pagination("Select a service:", choices, "Back to cluster selection")
 
     def select_service_action(self, service_name: str, task_info: list[TaskInfo]) -> str | None:
         choices = []
@@ -45,7 +46,7 @@ def select_service_action(self, service_name: str, task_info: list[TaskInfo]) ->
         choices.append({"name": "📋 Show service events", "value": "action:show_events"})
         choices.append({"name": "🚀 Force new deployment", "value": "action:force_deployment"})
 
-        return self.select_with_nav(
+        return select_with_auto_pagination(
             f"Select action for service '{service_name}':", choices, "Back to cluster selection"
         )
 

src/lazy_ecs/features/task/task.py

@@ -6,6 +6,7 @@
 
 from ...core.base import BaseAWSService
 from ...core.types import TaskDetails, TaskHistoryDetails, TaskInfo
+from ...core.utils import paginate_aws_list
 
 if TYPE_CHECKING:
     from mypy_boto3_ecs.client import ECSClient
@@ -19,8 +20,9 @@ def __init__(self, ecs_client: ECSClient) -> None:
         super().__init__(ecs_client)
 
     def get_tasks(self, cluster_name: str, service_name: str) -> list[str]:
-        response = self.ecs_client.list_tasks(cluster=cluster_name, serviceName=service_name)
-        return response.get("taskArns", [])
+        return paginate_aws_list(
+            self.ecs_client, "list_tasks", "taskArns", cluster=cluster_name, serviceName=service_name
+        )
 
     def get_task_info(self, cluster_name: str, service_name: str, desired_task_def_arn: str | None) -> list[TaskInfo]:
         task_arns = self.get_tasks(cluster_name, service_name)
@@ -57,27 +59,22 @@ def get_task_and_definition(
 
         return task, task_definition
 
+    def _list_tasks_paginated(self, cluster_name: str, service_name: str | None, desired_status: str) -> list[str]:
+        """List tasks with optional service name filtering."""
+        kwargs = {"cluster": cluster_name, "desiredStatus": desired_status}
+        if service_name:
+            kwargs["serviceName"] = service_name
+        return paginate_aws_list(self.ecs_client, "list_tasks", "taskArns", **kwargs)
+
     def get_task_history(self, cluster_name: str, service_name: str | None = None) -> list[TaskHistoryDetails]:
         """Get task history including stopped tasks with failure information."""
         task_arns = []
 
-        # Get running tasks
-        if service_name:
-            running_response = self.ecs_client.list_tasks(
-                cluster=cluster_name, serviceName=service_name, desiredStatus="RUNNING"
-            )
-        else:
-            running_response = self.ecs_client.list_tasks(cluster=cluster_name, desiredStatus="RUNNING")
-        task_arns.extend(running_response.get("taskArns", []))
+        running_arns = self._list_tasks_paginated(cluster_name, service_name, "RUNNING")
+        task_arns.extend(running_arns)
 
-        # Get stopped tasks
-        if service_name:
-            stopped_response = self.ecs_client.list_tasks(
-                cluster=cluster_name, serviceName=service_name, desiredStatus="STOPPED"
-            )
-        else:
-            stopped_response = self.ecs_client.list_tasks(cluster=cluster_name, desiredStatus="STOPPED")
-        task_arns.extend(stopped_response.get("taskArns", []))
+        stopped_arns = self._list_tasks_paginated(cluster_name, service_name, "STOPPED")
+        task_arns.extend(stopped_arns)
 
         if not task_arns:
             return []

src/lazy_ecs/features/task/ui.py

@@ -9,7 +9,7 @@
 from rich.table import Table
 
 from ...core.base import BaseUIComponent
-from ...core.navigation import add_navigation_choices
+from ...core.navigation import add_navigation_choices, select_with_auto_pagination
 from ...core.types import TaskDetails, TaskHistoryDetails
 from ...core.utils import print_warning, show_spinner
 from .task import TaskService
@@ -43,7 +43,7 @@ def select_task(self, cluster_name: str, service_name: str, desired_task_def_arn
 
         choices = [{"name": task["name"], "value": task["value"]} for task in available_tasks]
 
-        selected = self.select_with_nav("Select a task:", choices, "Back to service selection")
+        selected = select_with_auto_pagination("Select a task:", choices, "Back to service selection")
 
         if selected:
             console.print("Task selected successfully!", style="blue")
@@ -112,10 +112,8 @@ def select_task_feature(self, task_details: TaskDetails | None) -> str | None:
         if not containers:
             return None
 
-        # Build choices but don't add navigation - select_with_nav will handle it
         choices = []
 
-        # Add task-level features first - show task details as first option
         choices.extend(
             [
                 {"name": "Show task details", "value": "task_action:show_details"},
@@ -154,7 +152,7 @@ def select_task_feature(self, task_details: TaskDetails | None) -> str | None:
                 ]
             )
 
-        return self.select_with_nav("Select a feature for this task:", choices, "Back to service selection")
+        return select_with_auto_pagination("Select a feature for this task:", choices, "Back to service selection")
 
     def display_task_history(self, cluster_name: str, service_name: str) -> None:
         """Display task history with failure analysis."""

tests/conftest.py

@@ -0,0 +1,27 @@
+"""Shared pytest fixtures for tests."""
+
+from unittest.mock import Mock
+
+import pytest
+
+
+@pytest.fixture
+def mock_paginated_client():
+    """Create a mock AWS client with paginator support.
+
+    Returns a factory function that creates clients with specified pagination pages.
+
+    Example:
+        def test_something(mock_paginated_client):
+            pages = [{"clusterArns": ["arn1", "arn2"]}]
+            client = mock_paginated_client(pages)
+    """
+
+    def _create_client(pages: list[dict]) -> Mock:
+        client = Mock()
+        paginator = Mock()
+        paginator.paginate.return_value = pages
+        client.get_paginator.return_value = paginator
+        return client
+
+    return _create_client