632ac9e328
- Python CLI tool for rolling updates of CL-AppPipe-* and CL-SvcPipe-* stacks - Async update engine with configurable concurrency (asyncio.Semaphore) - Exponential backoff retry for API throttling - Dry-run mode for safe preview - IAM permission pre-validation - Comprehensive test suite (80 tests: 11 property-based + 69 unit) - Full spec documentation (requirements, design, tasks)
65 lines
2 KiB
Python
65 lines
2 KiB
Python
"""Stack discovery for the CloudFormation Stack Updater."""
|
|
|
|
from __future__ import annotations
|
|
|
|
from cfn_updater.config import NON_UPDATABLE_STATUSES
|
|
from cfn_updater.models import DiscoveredStack
|
|
|
|
|
|
# Statuses to exclude from list_stacks (DELETE_COMPLETE stacks are hidden by default
|
|
# but we explicitly exclude them to be safe).
|
|
_ACTIVE_STACK_STATUSES = [
|
|
"CREATE_IN_PROGRESS",
|
|
"CREATE_FAILED",
|
|
"CREATE_COMPLETE",
|
|
"ROLLBACK_IN_PROGRESS",
|
|
"ROLLBACK_FAILED",
|
|
"ROLLBACK_COMPLETE",
|
|
"DELETE_IN_PROGRESS",
|
|
"DELETE_FAILED",
|
|
"UPDATE_IN_PROGRESS",
|
|
"UPDATE_COMPLETE_CLEANUP_IN_PROGRESS",
|
|
"UPDATE_COMPLETE",
|
|
"UPDATE_FAILED",
|
|
"UPDATE_ROLLBACK_IN_PROGRESS",
|
|
"UPDATE_ROLLBACK_FAILED",
|
|
"UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS",
|
|
"UPDATE_ROLLBACK_COMPLETE",
|
|
"REVIEW_IN_PROGRESS",
|
|
"IMPORT_IN_PROGRESS",
|
|
"IMPORT_COMPLETE",
|
|
"IMPORT_ROLLBACK_IN_PROGRESS",
|
|
"IMPORT_ROLLBACK_FAILED",
|
|
"IMPORT_ROLLBACK_COMPLETE",
|
|
]
|
|
|
|
|
|
def discover_stacks(cfn_client, prefix: str) -> list[DiscoveredStack]:
|
|
"""List all stacks matching *prefix*. Paginates through all results.
|
|
|
|
Marks each stack as updatable or not based on its status against
|
|
``NON_UPDATABLE_STATUSES``.
|
|
"""
|
|
stacks: list[DiscoveredStack] = []
|
|
next_token: str | None = None
|
|
|
|
while True:
|
|
kwargs: dict = {"StackStatusFilter": _ACTIVE_STACK_STATUSES}
|
|
if next_token is not None:
|
|
kwargs["NextToken"] = next_token
|
|
|
|
response = cfn_client.list_stacks(**kwargs)
|
|
|
|
for summary in response.get("StackSummaries", []):
|
|
name = summary["StackName"]
|
|
if not name.startswith(prefix):
|
|
continue
|
|
status = summary["StackStatus"]
|
|
updatable = status not in NON_UPDATABLE_STATUSES
|
|
stacks.append(DiscoveredStack(name=name, status=status, updatable=updatable))
|
|
|
|
next_token = response.get("NextToken")
|
|
if not next_token:
|
|
break
|
|
|
|
return stacks
|