Last active
October 6, 2016 18:02
-
-
Save xuru/40f19e6a58c810929c6f83a8579835bf to your computer and use it in GitHub Desktop.
Generating Presigned Posts - https://boto3.readthedocs.io/en/latest/guide/s3.html#generating-presigned-posts
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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']) |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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