Skip to content

Instantly share code, notes, and snippets.

@brunsgaard
Created May 5, 2017 14:23
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 brunsgaard/9afb98f8928a658032bd4c5604f2c3ff to your computer and use it in GitHub Desktop.
Save brunsgaard/9afb98f8928a658032bd4c5604f2c3ff to your computer and use it in GitHub Desktop.
```python
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)
```
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment