Utilities#
This section provides detailed API documentation for utility modules that simplify common AWS CDK operations.
Deployment Utilities#
The deployment utilities module provides programmatic control over CDK stack deployments, making it easy to deploy, manage, and clean up CDK stacks without requiring a full CDK project structure.
Overview#
The Deployer class handles CDK CLI operations, CloudFormation stack management,
and lifecycle operations. It’s particularly useful for:
Functional Testing: Deploy and test stacks programmatically
CI/CD Pipelines: Automate stack deployments with custom logic
Dynamic Stack Management: Create and destroy stacks on-demand
Temporary Deployments: Use context managers for automatic cleanup
Key Features#
Automatic temporary directory management
Stack deployment tracking for cleanup
CloudFormation status monitoring
Stack output collection
Context manager support for automatic cleanup
Comprehensive error handling and logging
Timeout configuration for all operations
API Reference#
Basic Usage Example#
Simple deployment with automatic cleanup:
from aws_cdk import App
from core_aws_cdk.utils.deployment import Deployer
from core_aws_cdk.stacks.ecs.cluster.base import ECSClusterStack
# Create deployer with context manager for auto-cleanup
with Deployer(folder_name="test_deployment") as deployer:
# Create CDK app with deployer's output directory
app = App(outdir=str(deployer.cdk_out_dir))
# Define your stack
stack = ECSClusterStack(
app,
"TestCluster",
cluster_name="my-cluster",
enable_fargate=True
)
# Synthesize and deploy
deployer.synthesize(app)
result = deployer.deploy_stack("TestCluster")
if result.success:
print(f"Deployment successful in {result.duration_seconds:.2f}s")
print(f"Stack outputs: {result.outputs}")
else:
print(f"Deployment failed: {result.error}")
# Stack is automatically destroyed and temp dir cleaned up
Advanced Usage Example#
Manual deployment with status monitoring:
from aws_cdk import App, Environment
from core_aws_cdk.utils.deployment import Deployer
from core_aws_cdk.stacks.ecs.cluster.base import ECSClusterStack
# Initialize deployer
deployer = Deployer(
folder_name="ecs_deployment",
auto_approve=True,
region="us-east-1"
)
try:
# Create app
app = App(outdir=str(deployer.cdk_out_dir))
# Create stack
stack = ECSClusterStack(
app,
"ProductionCluster",
cluster_name="prod-cluster",
enable_fargate=True,
enable_container_insights=True,
env=Environment(
account="123456789012",
region="us-east-1"
)
)
# Synthesize
deployer.synthesize(app)
# List available stacks
stacks = deployer.list_stacks()
print(f"Available stacks: {stacks}")
# Show diff before deployment
diff = deployer.diff_stack("ProductionCluster")
print(f"Stack diff:\\n{diff}")
# Deploy with custom timeout
result = deployer.deploy_stack(
stack_name="ProductionCluster",
timeout_seconds=1200, # 20 minutes
context={"env": "production"}
)
if not result.success:
raise RuntimeError(f"Deployment failed: {result.error}")
# Wait for stack to be fully ready
deployer.wait_for_stack(
stack_name="ProductionCluster",
target_status="CREATE_COMPLETE",
max_wait_seconds=600
)
# Get stack status
status = deployer.get_stack_status("ProductionCluster")
print(f"Stack status: {status.status}")
print(f"Resources: {status.resources_count}")
print(f"Outputs: {status.outputs}")
finally:
# Manual cleanup
deployer.cleanup_all()
Functional Testing Example#
Using the deployer in functional tests:
import unittest
from aws_cdk import App
from core_aws_cdk.utils.deployment import Deployer
from core_aws_cdk.stacks.ecs.cluster.base import ECSClusterStack
class TestECSClusterDeployment(unittest.TestCase):
"""Functional tests that deploy real ECS clusters."""
def setUp(self) -> None:
"""Set up deployer before each test."""
self.deployer = Deployer(
folder_name=f"test_{self._testMethodName}",
auto_approve=True
)
def tearDown(self) -> None:
"""Clean up after each test."""
self.deployer.cleanup_all()
def test_fargate_cluster_deployment(self) -> None:
"""Test deploying a Fargate-only cluster."""
# Create app
app = App(outdir=str(self.deployer.cdk_out_dir))
# Create stack
stack = ECSClusterStack(
app,
"TestFargateCluster",
cluster_name="test-fargate-cluster",
enable_fargate=True,
fargate_capacity_providers=["FARGATE", "FARGATE_SPOT"]
)
# Synthesize and deploy
self.deployer.synthesize(app)
result = self.deployer.deploy_stack("TestFargateCluster")
# Assertions
self.assertTrue(result.success)
self.assertIsNotNone(result.outputs)
self.assertGreater(result.duration_seconds, 0)
# Verify stack status
status = self.deployer.get_stack_status("TestFargateCluster")
self.assertEqual(status.status, "CREATE_COMPLETE")
self.assertTrue(status.exists)
if __name__ == "__main__":
unittest.main()
Multi-Stack Deployment Example#
Deploying multiple dependent stacks:
from aws_cdk import App, Environment
from core_aws_cdk.utils.deployment import Deployer
from core_aws_cdk.stacks.network.base import NetworkStack
from core_aws_cdk.stacks.ecs.cluster.base import ECSClusterStack
deployer = Deployer(folder_name="multi_stack")
try:
app = App(outdir=str(deployer.cdk_out_dir))
env = Environment(account="123456789012", region="us-east-1")
# Create network stack
network_stack = NetworkStack(app, "NetworkStack", env=env)
vpc = network_stack.create_vpc(
vpc_id="MainVPC",
cidr="10.0.0.0/16",
max_azs=2
)
# Create ECS cluster stack
cluster_stack = ECSClusterStack(
app,
"ClusterStack",
cluster_name="app-cluster",
vpc=vpc,
enable_fargate=True,
env=env
)
# Synthesize
deployer.synthesize(app)
# Deploy in order
network_result = deployer.deploy_stack("NetworkStack")
if not network_result.success:
raise RuntimeError("Network stack failed")
cluster_result = deployer.deploy_stack("ClusterStack")
if not cluster_result.success:
raise RuntimeError("Cluster stack failed")
print("All stacks deployed successfully")
except Exception as e:
print(f"Deployment error: {e}")
deployer.cleanup_all()
Deployment Result Data Classes#
DeploymentResult#
Returned by deploy_stack() with deployment information:
@dataclass
class DeploymentResult:
stack_name: str # Name of the deployed stack
success: bool # Whether deployment succeeded
outputs: Dict[str, str] # CloudFormation outputs
error: Optional[str] # Error message if failed
duration_seconds: float # Deployment duration
StackStatus#
Returned by get_stack_status() with current stack information:
@dataclass
class StackStatus:
stack_name: str # Name of the stack
status: str # CloudFormation status
exists: bool # Whether stack exists
resources_count: int # Number of resources
outputs: Dict[str, str] # Stack outputs
Common CloudFormation Status Values#
When monitoring stack operations, you’ll encounter these status values:
Creation Statuses:
CREATE_IN_PROGRESS- Stack is being createdCREATE_COMPLETE- Stack created successfullyCREATE_FAILED- Stack creation failed
Update Statuses:
UPDATE_IN_PROGRESS- Stack is being updatedUPDATE_COMPLETE- Stack updated successfullyUPDATE_FAILED- Stack update failedUPDATE_ROLLBACK_IN_PROGRESS- Update failed, rolling backUPDATE_ROLLBACK_COMPLETE- Rollback completed
Deletion Statuses:
DELETE_IN_PROGRESS- Stack is being deletedDELETE_COMPLETE- Stack deleted successfullyDELETE_FAILED- Stack deletion failed
Best Practices#
Use Context Managers: Always use the
withstatement for automatic cleanup:with Deployer(folder_name="deployment") as deployer: # Your deployment code pass # Automatic cleanup happens here
Set Appropriate Timeouts: ECS clusters may take 10+ minutes to deploy:
result = deployer.deploy_stack( "ECSStack", timeout_seconds=1200 # 20 minutes )
Monitor Stack Status: Use
wait_for_stack()for operations that need verification:deployer.wait_for_stack( "MyStack", "CREATE_COMPLETE", max_wait_seconds=600 )
Handle Errors Gracefully: Always check deployment results:
result = deployer.deploy_stack("MyStack") if not result.success: logger.error(f"Deployment failed: {result.error}") deployer.cleanup_all() return
Use Explicit Regions: Specify region to avoid relying on AWS profile:
deployer = Deployer( folder_name="deployment", region="us-east-1" # Explicit region )
Clean Up Resources: Always ensure cleanup happens, even on errors:
deployer = Deployer(folder_name="test") try: # Your deployment code pass finally: deployer.cleanup_all()
Troubleshooting#
CDK Synthesis Fails
If synthesize() fails, check:
CDK app is properly configured with
outdir=str(deployer.cdk_out_dir)All stack parameters are valid
Required AWS permissions are available
Deployment Times Out
If deployment exceeds timeout:
Increase
timeout_secondsparameter (ECS clusters need 10-15 minutes)Check AWS CloudFormation console for stack progress
Verify AWS service limits haven’t been exceeded
Stack Not Found After Deployment
If get_stack_status() returns NOT_FOUND:
Verify correct stack name (case-sensitive)
Check correct AWS region is configured
Confirm deployment actually succeeded
Cleanup Fails
If cleanup_all() has errors:
Check CloudFormation console for stack status
Verify stacks don’t have resources that prevent deletion
Manually delete stacks via AWS console if needed