Skip to content

Instantly share code, notes, and snippets.

@cwilkes
Created March 19, 2017 05:12
Show Gist options
  • Save cwilkes/b630ee1f198bc6cb6fba3e4a6ce41285 to your computer and use it in GitHub Desktop.
Save cwilkes/b630ee1f198bc6cb6fba3e4a6ce41285 to your computer and use it in GitHub Desktop.
python code to insert into dynamo
from __future__ import print_function
import boto3
import json
import uuid
from decimal import Decimal
from datetime import datetime
import random
print('Loading function')
audio = ('Lorem Ipsum is simply dummy text of the printing and typesetting industry. '
'Lorem Ipsum has been the industry\'s standard dummy text ever since the 1500s, '
'when an unknown printer took a galley of type and scrambled it to make a type specimen book. '
'It has survived not only five centuries, but also the leap into electronic typesetting, '
'remaining essentially unchanged. It was popularised in the 1960s with the release of '
'Letraset sheets containing Lorem Ipsum passages, and more recently with desktop publishing '
'software like Aldus PageMaker including versions of Lorem Ipsum').replace(',', ' ').split()
usernames = ['jason', 'brey', 'datboi', 'marc', 'gavin', 'urmom', 'urmom']
type_of_user = ['coordinates',
'design',
'contractor',
'owner',
'surgeon',
'nurse',
'coordinates']
headers=[
'GUID',
'x',
'y',
'z',
'spherical image',
'audio',
'audio transcription',
'username',
'design discipline affected',
'type of user',
'estimated cost of change',
'estimated time of design change',
'level of impact']
COL_INTS = set(['estimated cost of change', 'estimated time of design change', 'level of impact'])
COL_FLOATS = set(['x', 'y', 'z'])
def respond(err, res=None):
return {
'statusCode': '400' if err else '200',
'body': err.message if err else json.dumps(res),
'headers': {
'Content-Type': 'application/json',
},
}
def csv_respond(err, data=None):
ret=['"%s"'%('","'.join(headers)), ]
ret.extend([','.join(map(str, _)) for _ in data])
return {
'statusCode': '400' if err else '200',
'body': err.message if err else '\r\n'.join(ret) + '\r\n',
'headers': {
'Content-Type': 'text/jsv',
},
}
table_name = 'bar'
bucket_name = 'saywhat1'
s3_url = 'https://s3.amazonaws.com/{}'.format(bucket_name)
def make_db_entry(payload, names):
ret = dict()
for name in names:
ret[name] = Decimal(str(payload[name]))
ret['id'] = uuid.uuid4().hex
return ret
def upload_audio(data, file_id):
s3 = boto3.resource('s3').Bucket(bucket_name)
x= s3.put_object(Key='audio/%s.wav' % (file_id,))
return 'https://s3.amazonaws.com/%s/%s' % (bucket_name, x.key)
def lambda_handler(event, context):
print('event', event)
dynamo = boto3.resource('dynamodb').Table(table_name)
operation = event['httpMethod']
if operation == 'POST':
payload = json.loads(event['body'])
pay2 = make_db_entry(payload, ('x', 'y', 'z'))
for key in headers[4:-3]:
pay2[key] = uuid.uuid4().hex
for key in headers[-3:]:
pay2[key] = random.randint(0,5)
pay2['username'] = random.choice(usernames)
words = ' '.join([random.choice(audio) for _ in range(random.randint(3,5))])
pay2['audio transcription'] = words
pay2['spherical image'] = 'http://sphcst.com/%s' % (pay2['id'], )
pay2['audio'] = 'https://s3.amazonaws.com/%s/%s.wav' % (bucket_name, pay2['id'])
pay2['type of user'] = random.choice(type_of_user)
pay2['timestamp'] = datetime.now().strftime('%Y-%m-%d %I:%M:%S %p')
return respond(None, dynamo.put_item(Item=pay2))
elif operation == 'GET':
ret = []
# need to convert the odd "Decimal" type into a decimal
# this is going to return CSV
for item in dynamo.scan()['Items']:
def g(name):
return float(item[name])
q = [item['id'],g('x'),g('y'),g('z'),]
q.extend([item.get(_, '') for _ in headers[4:]])
# item 6 are words that could have a comma
q[6] = '"%s"' % (q[6], )
# convert the tuple into a CSV string
ret.append(q)
return csv_respond(None, ret)
else:
return respond(ValueError('Unsupported method "{}"'.format(operation)))
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment