saturncloud_kinesis_producer.py

import calendar
import json
import os
import random
import time
import boto3

from datetime import datetime


AWS_ACCESS_KEY_ID = os.getenv('AWS_ACCESS_KEY_ID')
AWS_SECRET_ACCESS_KEY = os.getenv('AWS_SECRET_ACCESS_KEY')
AWS_DEFAULT_REGION = os.getenv('AWS_DEFAULT_REGION')
AWS_APP_STREAM_NAME = os.getenv('AWS_APP_STREAM_NAME')

kinesis_client = boto3.client('kinesis', aws_access_key_id=AWS_ACCESS_KEY_ID,
                              aws_secret_access_key=AWS_SECRET_ACCESS_KEY,
                              region_name=AWS_DEFAULT_REGION)


def put_to_stream(partition_key, property_value, property_timestamp):
    payload = {
        'prop': str(property_value),
        'timestamp': str(property_timestamp),
        'PartitionKey': partition_key
    }
    print(payload)
    # PartitionKey if same partition key is used it data will be inserted into 1 shard
    # remember more shards your cost will increase
    put_response = kinesis_client.put_record(
        StreamName=AWS_APP_STREAM_NAME,
        Data=json.dumps(payload),
        PartitionKey=partition_key)
    print(put_response)

    
# simulate adding alot of data into the stream
while True:
    property_value = random.randint(40, 120)
    property_timestamp = calendar.timegm(datetime.utcnow().timetuple())
    partition_key = 'timothy.mugayi'
    put_to_stream(partition_key, property_value, property_timestamp)
    # wait for 5 second
    time.sleep(5)