Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
Serverless function snippet to return an S3 file
def get(event, context):
response = _get_error_responce('Incorrect URL format: Use /foobar/<a>/<b>/?type=<url/file>')
if 'queryStringParameters' in event and 'pathParameters' in event:
if event['pathParameters'] and 'a' in event['pathParameters'] and 'b' in event['pathParameters']:
a = event['pathParameters']['a']
b = event['pathParameters']['b']
s3_key = 's3key/to/file/{a}/{b}'.format(a=a, b=b)
if event['queryStringParameters'] and 'type' in event['queryStringParameters']:
type = event['queryStringParameters']['type']
if type == 'file':
try:
obj_ref = s3_resource.Object(bucket_name=BUCKET, key=s3_key)
except Exception as e:
obj_ref = None
response = _get_error_responce(e)
if obj_ref:
try:
obj = obj_ref.get()
response = {
'statusCode': 200,
'headers': {
'Content-Type': obj['ContentType']
},
'body': base64.encodebytes(obj['Body'].read()).decode(),
'isBase64Encoded': True
}
except Exception as e:
response = _get_error_responce(e)
elif type == 'url':
try:
url = s3_client.generate_presigned_url('get_object', Params={
'Bucket': BUCKET,
'Key': s3_key
}, ExpiresIn=100)
except Exception as e:
url = None
response = _get_error_responce(e)
if url:
response = {
"statusCode": 200,
"body": json.dumps({
"url": url
})
}
else:
response = _get_error_responce('Illegal type: {}'.format(type))
return response
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment