Skip to content

Instantly share code, notes, and snippets.

@henprtz
Last active Jun 5, 2018
Embed
What would you like to do?
from json import dumps
from uuid import uuid4
import boto3
from time import time
from os import environ
KINESIS_STREAM_VAR = 'KINESIS_STREAM_NAME'
client = boto3.client('kinesis')
def place_vote(event, context):
"""
Reading vote information from user and publishing it to the kinesis stream
:param event: event object
:param context: event context
:return: response object
"""
data = event['pathParameters']
record_data = {
'time_id': time(),
'vote_id': data['vote_id'],
'token': data['token'],
'vote_choice': data['vote_choice']
}
result = client.put_record(
StreamName=environ(KINESIS_STREAM_VAR),
Data=dumps(record_data),
PartitionKey=str(uuid4())
)
response = {
"statusCode": 200,
"body": dumps(result)
}
return response
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment