localstack

import subprocess import time import os import boto3 def start_localstack(): print('Starting LocalStack in the background...') # Use subprocess to run 'localstack start' non-blocking # Note: Requires 'localstack' CLI to be in PATH process = subprocess.Popen(['localstack', 'start'], stdout=subprocess.PIPE, stderr=subprocess.PIPE, text=True) # Give LocalStack some time to start up time.sleep(15) # Check if LocalStack is running via 'localstack status' status_output = subprocess.run(['localstack', 'status'], capture_output=True, text=True) if 'running' in status_output.stdout: print('LocalStack is running.') else: print('LocalStack failed to start or is not running correctly.') print('STDOUT:', status_output.stdout) print('STDERR:', status_output.stderr) # Optionally, terminate the Popen process if it didn't start correctly # process.terminate() # return None return process def stop_localstack(process): print('Stopping LocalStack...') subprocess.run(['localstack', 'stop']) if process: # Terminate the Popen process if it was started this way process.terminate() process.wait() print('LocalStack stopped.') def main(): localstack_process = None try: localstack_process = start_localstack() if not localstack_process: return # Now, interact with a service (e.g., S3) using boto3 print('\nInteracting with S3 on LocalStack...') s3_client = boto3.client( 's3', region_name='us-east-1', endpoint_url='http://localhost:4566', aws_access_key_id='test', aws_secret_access_key='test' ) bucket_name = 'my-local-test-bucket' try: s3_client.create_bucket(Bucket=bucket_name) print(f"Bucket '{bucket_name}' created successfully.") response = s3_client.list_buckets() print('Existing buckets:') for bucket in response['Buckets']: print(f" - {bucket['Name']}") except Exception as e: print(f"Error interacting with S3: {e}") finally: stop_localstack(localstack_process) if __name__ == '__main__': # Ensure Docker is running before executing this script # On first run, 'localstack start' will download necessary Docker images. main()