Skip to content

Instantly share code, notes, and snippets.

@xuru
Last active October 6, 2016 18:02
Show Gist options
  • Star 0 You must be signed in to star a gist
  • Fork 0 You must be signed in to fork a gist
  • Save xuru/40f19e6a58c810929c6f83a8579835bf to your computer and use it in GitHub Desktop.
Save xuru/40f19e6a58c810929c6f83a8579835bf to your computer and use it in GitHub Desktop.
from flask_restful import Resource
class UserProfileImageResource(Resource):
@jwt_required()
def post(self, user_id):
"""
/v1/users/<user_id>/signed_post
"""
args = request.json
if 'Content-Type' not in args:
raise APIException("Content-Type must be specified")
content_type = args.get('Content-Type')
ext = mimetypes.guess_extension(content_type)
if not ext:
raise APIException("Unable to get extension for Content-Type")
user = User.get_by_id(user_id)
if not user:
raise APIException("User not found or unauthorized")
s3 = S3(current_app) # S3 helper functions
identity = user.identity.hex
key = identity + ext
# check if it exists
keys = s3.list_keys(prefix=key)
if keys:
s3.delete(key)
post_info = get_upload_info(key)
return flask.jsonify(method='POST', url=post_info['url'], fields=post_info['fields'])
import mimetypes
import boto3
def get_upload_info(key):
conditions = []
fields = {'acl': "private"}
content_type = mimetypes.guess_type(key)[0]
if content_type:
fields['Content-Type'] = content_type
# Boto doesn't automatically add fields to conditions.
for field_key, field_value in fields.items():
conditions.append({field_key: field_value})
client = boto3.client('s3', aws_access_key_id=AWS_ACCESS_KEY, aws_secret_access_key=AWS_SECRET_ACCESS_KEY)
post_info = client.generate_presigned_post(
Bucket=S3_BUCKET,
Key=key,
Fields=fields,
Conditions=conditions,
ExpiresIn=600
)
return post_info
import json
from pprint import pprint
import logging
import requests
session = requests.Session()
session.headers.update({'Content-Type': 'application/json'})
def login():
r = session.post(api_url + '/auth', data=json.dumps(auth_data))
r.raise_for_status()
json_args = r.json()
token = json_args['data']['token']
session.headers.update({'Authorization': 'JWT {}'.format(token)})
return json_args['data']['user']['identity']
def get_upload_info(user_id):
# Get the permissions and url
resp = session.post(
api_url + "/v0/users/{}/signed_post".format(user_id),
data=json.dumps({'Content-Type': 'image/png'})
)
resp.raise_for_status()
return resp.json()
def upload_image_to_s3(data):
url = data['url']
fields = data['fields']
filename = fields['key']
image_data = open("/Users/xxxxx/Pictures/avatar/overlord59-magic-tux.png", 'rb').read()
resp = requests.post(url, data=fields, files={'file': image_data})
resp.raise_for_status() # BOOM
pprint(resp.text)
identity = login()
data = get_upload_info(identity)
pprint(data)
upload_image_to_s3(data)
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment