Fix: Handle AWS ECS 10-service limit in describe_services

pyproject.toml

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

src/lazy_ecs/core/utils.py

@@ -8,7 +8,7 @@
 import threading
 from collections.abc import Iterator
 from contextlib import contextmanager, suppress
-from typing import TYPE_CHECKING, Literal
+from typing import TYPE_CHECKING, Any, Literal
 
 from rich.console import Console
 from rich.spinner import Spinner
@@ -80,6 +80,12 @@ def show_spinner() -> Iterator[None]:
         yield
 
 
+def batch_items(items: list[Any], batch_size: int) -> Iterator[list[Any]]:
+    """Split a list into batches of specified size."""
+    for i in range(0, len(items), batch_size):
+        yield items[i : i + batch_size]
+
+
 def paginate_aws_list(
     client: ECSClient,
     operation_name: Literal[

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, paginate_aws_list
+from ...core.utils import batch_items, determine_service_status, extract_name_from_arn, paginate_aws_list
 
 if TYPE_CHECKING:
     from typing import Any
@@ -31,9 +31,13 @@ def get_service_info(self, cluster_name: str) -> list[ServiceInfo]:
         if not service_names:
             return []
 
-        response = self.ecs_client.describe_services(cluster=cluster_name, services=service_names)
-        services = response.get("services", [])
-        return [_create_service_info(service) for service in services]
+        all_services = [
+            service
+            for batch in batch_items(service_names, 10)
+            for service in self.ecs_client.describe_services(cluster=cluster_name, services=batch).get("services", [])
+        ]
+
+        return [_create_service_info(service) for service in all_services]
 
     def get_desired_task_definition_arn(self, cluster_name: str, service_name: str) -> str | None:
         response = self.ecs_client.describe_services(cluster=cluster_name, services=[service_name])

tests/test_aws_service.py

@@ -268,6 +268,32 @@ def test_get_service_info(ecs_client_with_services) -> None:
         assert "pending_count" in info
 
 
+def test_get_service_info_with_more_than_10_services():
+    with mock_aws():
+        client = boto3.client("ecs", region_name="us-east-1")
+        client.create_cluster(clusterName="production")
+
+        client.register_task_definition(
+            family="app-task",
+            containerDefinitions=[{"name": "app", "image": "nginx", "memory": 256}],
+        )
+
+        for i in range(15):
+            client.create_service(
+                cluster="production",
+                serviceName=f"service-{i:02d}",
+                taskDefinition="app-task",
+                desiredCount=2,
+            )
+
+        service = ECSService(client)
+        service_info = service.get_service_info("production")
+
+        assert len(service_info) == 15
+        service_names = {info["name"] for info in service_info}
+        assert all(any(f"service-{i:02d}" in name for name in service_names) for i in range(15))
+
+
 def test_get_tasks(ecs_client_with_tasks) -> None:
     service = ECSService(ecs_client_with_tasks)
     tasks = service.get_tasks("production", "web-api")

tests/test_utils.py

@@ -0,0 +1,39 @@
+"""Tests for utility functions."""
+
+from lazy_ecs.core.utils import batch_items
+
+
+def test_batch_items_basic():
+    batches = list(batch_items([1, 2, 3, 4, 5, 6, 7, 8, 9, 10], 3))
+
+    assert batches == [[1, 2, 3], [4, 5, 6], [7, 8, 9], [10]]
+
+
+def test_batch_items_exact_fit():
+    batches = list(batch_items([1, 2, 3, 4, 5, 6], 3))
+
+    assert batches == [[1, 2, 3], [4, 5, 6]]
+
+
+def test_batch_items_single_batch():
+    batches = list(batch_items([1, 2, 3], 10))
+
+    assert batches == [[1, 2, 3]]
+
+
+def test_batch_items_empty_list():
+    batches = list(batch_items([], 5))
+
+    assert batches == []
+
+
+def test_batch_items_size_one():
+    batches = list(batch_items([1, 2, 3, 4, 5], 1))
+
+    assert batches == [[1], [2], [3], [4], [5]]
+
+
+def test_batch_items_strings():
+    batches = list(batch_items(["a", "b", "c", "d", "e", "f", "g"], 3))
+
+    assert batches == [["a", "b", "c"], ["d", "e", "f"], ["g"]]