Last active
June 25, 2024 18:17
-
-
Save Anishmourya/b5006c3f282cb02b0048b7d0df1757e3 to your computer and use it in GitHub Desktop.
FLASK + S3 Presigned URLs + FIle upload to s3
This file contains 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 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() |
This file contains 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
id_key="####" | |
secret_key="####" | |
bucket_name="####" |
This file contains 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 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