Skip to content

Instantly share code, notes, and snippets.

@Anishmourya
Last active June 26, 2023 18:42
Show Gist options
  • Save Anishmourya/b5006c3f282cb02b0048b7d0df1757e3 to your computer and use it in GitHub Desktop.
Save Anishmourya/b5006c3f282cb02b0048b7d0df1757e3 to your computer and use it in GitHub Desktop.
FLASK + S3 Presigned URLs + FIle upload to s3
from s3 import AwsS3UploadClass
from config import id_key
from config import secret_key
from config import bucket_name
from flask import Flask
from flask import jsonify
from flask import request
import requests
app = Flask(__name__)
@app.route('/upload_file', methods=["POST"])
def upload_file():
if request.method == "POST":
file = None
if "file" in request.files:
file = request.files['file']
else:
return jsonify(error="requires file")
# import s3 upload class
s3 = AwsS3UploadClass(id_key,secret_key,bucket_name)
# save file
key = "file_name"
file.save(key)
# generate presgined post url
response = s3.create_presigned_post(key)
if response is None:
return jsonify(error="key cannot None")
files = [
('file', open(key, 'rb'))
]
upload_response = requests.post(response['url'], data=response['fields'], files=files)
if upload_response.status_code == 204:
# remove file
os.remove(key)
return jsonify("file successfully uploaded to s3")
if __name__ == '__main__':
app.run()
id_key="####"
secret_key="####"
bucket_name="####"
import logging
import boto3
from botocore.exceptions import ClientError
class AwsS3UploadClass(object):
def __init__(self, id_key, secret_key, bucket_name):
"""
AWS s3 Upload Class
Arguments:
* id_key (string) aws access key id
* secret_key (string) aws secret access key
* bucket (string) the S3 bucket to connect to
Attributes:
* bucket (S3.Bucket) the bucket instance for the specified
`bucket_name`
"""
self.id_key = id_key
self.secret_key = secret_key
self.bucket_name = bucket_name
self.client = boto3.client('s3',
endpoint_url=None,
aws_access_key_id=self.id_key,
aws_secret_access_key=self.secret_key,
region_name='eu-west-2'
)
def get_bucket(self, bucket_name):
try:
bucket = self.client.get_bucket(bucket_name)
except ClientError as e:
bucket = None
return bucket
def create_presigned_post(self, object_name,
fields=None, conditions=None, expiration=3600):
"""Generate a presigned URL S3 POST request to upload a file
:param bucket_name: string
:param object_name: string
:param fields: Dictionary of prefilled form fields
:param conditions: List of conditions to include in the policy
:param expiration: Time in seconds for the presigned URL to remain valid
:return: Dictionary with the following keys:
url: URL to post to
fields: Dictionary of form fields and values to submit with the POST
:return: None if error.
"""
s3_client = boto3.client('s3')
try:
response = s3_client.generate_presigned_post(self.bucket_name,
object_name,
Fields=fields,
Conditions=conditions,
ExpiresIn=expiration)
except ClientError as e:
logging.error(e)
return None
# The response contains the presigned URL and required fields
return response
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment