localstack-test.py

import localstack_client.session as boto3
import requests
import os

# Check the health of the service.
# If the service is running, return True.
def check_health(service) -> bool:
    # https://docs.localstack.cloud/references/internal-endpoints/
    endpoint = 'http://localhost:4566/_localstack/health'
    response = requests.get(endpoint)
    json_response = response.json()
    if json_response['services'][service] == 'running':
        return True
    return False

def bucket_exists(s3, bucket_name: str) -> bool:
    try:
        s3.head_bucket(Bucket=bucket_name)
        return True
    except s3.exceptions.ClientError as e:
        # If a client error is thrown, then check that it was a 404 error.
        # If it was a 404 error, then the bucket does not exist.
        error_code = e.response['Error']['Code']
        if error_code == '404':
            return False

        # Otherwise, raise the error.
        raise

def list_buckets(s3):
    """Generator that yields S3 bucket names."""
    response = s3.list_buckets()
    for bucket in response.get('Buckets', []):
        yield bucket['Name']

def stack_exists(cf, stack_name: str) -> bool:
    """Check if a CloudFormation stack exists."""
    
    try:
        cf.describe_stacks(StackName=stack_name)
        return True  # Stack exists
    except cf.exceptions.ClientError as e:
        if "does not exist" in str(e):
            return False  # Stack does not exist
        raise  # Re-raise other exceptions

def create_stack(cf, stack_name: str, template_body: str, overwrite: bool = False) -> bool:
    """Create a CloudFormation stack."""

    if not stack_exists(cf, stack_name):
        cf.create_stack(StackName=stack_name, TemplateBody=template_body)
        return True
    
    if not overwrite:
        return False

    cf.update_stack(StackName=stack_name, TemplateBody=template_body)

def load_template(stack_name: str) -> str:
    """Load a CloudFormation template."""
    template = os.path.join(os.path.dirname(__file__), 'cdk.out', f'{stack_name}.template.json')
    with open(template, 'r') as f:
        return f.read()
    
def set_bootstrap_template_version(ssm):
    # https://stackoverflow.com/a/74122292/8341513
    ssm.put_parameter(
        Name='/cdk-bootstrap/hnb659fds/version',
        Value='20',
        Type='String',
        Overwrite=True
    )

def list_instances(ec2):
    response = ec2.describe_instances()

    for reservation in response.get('Reservations', []):
        for instance in reservation.get('Instances', []):
            yield instance['InstanceId']

if __name__ == '__main__':
    s3 = boto3.client('s3')
    cf = boto3.client('cloudformation')
    ssm = boto3.client('ssm')
    ec2 = boto3.client('ec2')

    # set_bootstrap_template_version(ssm)
    # template = load_template('InfrastructureStack')
    # create_stack(cf, 'InfrastructureStack', template)
    # print(stack_exists(cf, 'InfrastructureStack'))

    # list_instances(ec2)

    for bucket in list_buckets(s3):
        print(bucket)