types-aiobotocore-kinesis

import asyncio from aiobotocore.session import get_session from types_aiobotocore_kinesis.client import KinesisClient from types_aiobotocore_kinesis.type_defs import PutRecordRequestRequestTypeDef import os async def put_kinesis_record(stream_name: str, data: str, partition_key: str): session = get_session() async with session.create_client("kinesis", region_name=os.environ.get('AWS_REGION', 'us-east-1')) as client: # type: KinesisClient request_params: PutRecordRequestRequestTypeDef = { "StreamName": stream_name, "Data": data.encode('utf-8'), "PartitionKey": partition_key, } response = await client.put_record(**request_params) print(f"Shard ID: {response['ShardId']}, Sequence Number: {response['SequenceNumber']}") async def main(): # Example usage: Replace with your stream name, data, and partition key # Ensure AWS_REGION environment variable is set or passed to create_client # Also ensure AWS credentials are configured. await put_kinesis_record("my-test-stream", "Hello Kinesis!", "my-partition-key") if __name__ == "__main__": # For demonstration, setting dummy env var if not present if 'AWS_REGION' not in os.environ: os.environ['AWS_REGION'] = 'us-east-1' asyncio.run(main())