def send_to_kinesis_stream(events, stream_name, partition_key=None,
                           packer=None, serializer=json.dumps):
    """Sends events to a Kinesis stream."""
    if not events:
        logger.info("No events provided: nothing delivered to Firehose")
        return

    records = []
    for event in events:
        if not partition_key:
            partition_key_value = str(uuid.uuid4())
        elif hasattr(partition_key, "__call__"):
            partition_key_value = partition_key(event)
        else:
            partition_key_value = partition_key

        if not isinstance(event, str):
            event = serializer(event)

        if packer:
            event = packer(event)

        record = {"Data": event,
                  "PartitionKey": partition_key_value}
        records.append(record)

    kinesis = boto3.client("kinesis")
    resp = kinesis.put_records(StreamName=stream_name, Records=records)
    return resp
def send_to_kinesis_stream ( events , stream_name , partition_key = None , packer = None , serializer = json . dumps ) : if not events : logger . info ( "No events provided: nothing delivered to Firehose" ) return records = [ ] for event in events : if not partition_key : partition_key_value = str ( uuid . uuid4 ( ) ) elif hasattr ( partition_key , "__call__" ) : partition_key_value = partition_key ( event ) else : partition_key_value = partition_key if not isinstance ( event , str ) : event = serializer ( event ) if packer : event = packer ( event ) record = { "Data" : event , "PartitionKey" : partition_key_value } records . append ( record ) kinesis = boto3 . client ( "kinesis" ) resp = kinesis . put_records ( StreamName = stream_name , Records = records ) return resp
Sends events to a Kinesis stream.