localstack_client_s3_bucket_and_kinesis_stream_quickstart.py

1import localstack_client.session
2
3# Create a session that automatically points to LocalStack endpoints
4session = localstack_client.session.Session()
5
6# Create an S3 client and a bucket
7s3 = session.client('s3')
8s3.create_bucket(Bucket='test-bucket')
9
10# List buckets to verify
11response = s3.list_buckets()
12print("Buckets:", [b['Name'] for b in response.get('Buckets', [])])
13
14# Create a Kinesis client and a stream
15kinesis = session.client('kinesis')
16stream_name = 'test-stream'
17kinesis.create_stream(StreamName=stream_name, ShardCount=1)
18
19# List streams to verify
20response = kinesis.list_streams()
21print("Streams:", response.get('StreamNames', []))