s3_flask_python.py

import threading

import boto3
import flask


local = threading.local()


class S3:
    """
    Flask extension to return a boto S3 client.
    """

    def __init__(self, app=None):
        self.app = app
        if app is not None:
            self.init_app(app)

    def init_app(self, app):
        pass

    def _client(self):
        if not hasattr(local, 's3'):
            config = flask.current_app.config
            session = boto3.session.Session(
                aws_access_key_id=config['AWS_ACCESS_KEY_ID'],
                aws_secret_access_key=config['AWS_SECRET_ACCESS_KEY'],
            )
            local.s3 = session.client('s3')
        return local.s3

    def __getattr__(self, item):
        return getattr(self._client(), item)


if __name__ == '__main__':

    from flask import Flask
    app = Flask(__name__)
    s3 = S3(app)

    @app.route("/put")
    def put():
        ident = str(threading.get_ident())
        s3.put_object(
            Bucket='vml-global-smartscan',
            Key=ident,
            Body=ident.encode()*5000000)
        return ident

    @app.route("/get")
    def hello():
        ident = str(threading.get_ident())
        return s3.get_object(
            Bucket='vml-global-smartscan',
            Key=ident)['Body'].read()

    if __name__ == "__main__":
        app.run(debug=True, threaded=True)