Leveraging boto with Google App Engine allows access to S3 (and many other boto integrations)
download the boto source and copy it into your app engine project
# Boto and Google App Engine error: E ImportError: No module named boto.pyami.config
# workaround solution (due to how Boto looks for some secret key config), move boto to the app engine project root level
# /myproject/app.yaml , main.py, etc.
# /myproject/handlers
# /myproject/boto
from boto.sts import STSConnection
from boto.s3.connection import S3Connection
class MyService( object ):
# ...
def get_s3_upload_url( self , fileHash, sizeInBytes, s3Key, s3Secret, s3Bucket ):
sts_conn = STSConnection( s3Key , s3Secret )
temp_session = sts_conn.get_session_token( 900 ) # TODO: memcache (generating the session takes 1 second!)
temp_connection = S3Connection( temp_session.access_key , temp_session.secret_key , security_token = temp_session.session_token )
headers = {'Content-Length': str( sizeInBytes ) }
name = '-'.join([ fileHash, str( sizeInBytes ) ])
URL = temp_connection.generate_url( expires_in=300, method='PUT', bucket=s3Bucket, key=name, headers=headers ) # if set to 'GET' and without headers can be used for download
return URL
# TODO: pass in the temp_session? (dependency injection?)
def get_s3_download_url( self , fileHash, sizeInBytes, s3Key, s3Secret, s3Bucket ):
sts_conn = STSConnection( s3Key , s3Secret )
temp_session = sts_conn.get_session_token( 900 ) # TODO: memcache (generating the session takes 1 second!)
temp_connection = S3Connection( temp_session.access_key , temp_session.secret_key , security_token = temp_session.session_token )
name = '-'.join([ fileHash, str( sizeInBytes ) ])
URL = temp_connection.generate_url( expires_in=300, method='GET', bucket=s3Bucket, key=name )
return URL
Boto and Google App Engine error: E ImportError: No module named boto.sts make sure that for your tests you run an external script which can import it to the python path
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 | #!/bin/sh
# pytest.sh
# make sure you have the requests library installed ( pip install requests ) for INT tests
GAE_HOME=/usr/local/google_appengine
export GAE_HOME
GAE_WEBAPP2=`ls -dt1 ${GAE_HOME}/lib/webapp2* | head -n1`
export GAE_WEBAPP2
GAE_JINJA2=`ls -dt1 ${GAE_HOME}/lib/jinja2* | head -n1`
export GAE_WEBAPP2
PYTHONPATH=${GAE_HOME}:${GAE_HOME}/lib/yaml/lib:${GAE_HOME}/lib/antlr3:${GAE_HOME}/lib/simplejson:${GAE_HOME}/lib/fancy_urllib:${GAE_WEBAPP2}:${GAE_JINJA2}
# assuming the project directory is one above the appengine and test directories (i.e. myproject/appengine , myproject/unittests, myproject/integrationtests
REPO_HOME=..
export REPO_HOME
PYTHONPATH=${PYTHONPATH}:${REPO_HOME}/appengine:${REPO_HOME}/unittests:${REPO_HOME}/integrationtests
export PYTHONPATH
PYTHONPATH=${PYTHONPATH}:${REPO_HOME}/appengine/boto
py.test $*
|
AppEngineFixture.py
# -*- coding: utf-8 -*-
import pytest
from google.appengine.ext import testbed # Make sure GAE libraries are in your PYTHONPATH , https://developers.google.com/appengine/docs/python/tools/localunittesting
from google.appengine.ext.webapp import Request, Response # Make sure GAE libraries are in your PYTHONPATH
# Huge hack to override google app engine test framework not initializing correctly
APP_ID = 'myapp'
import os
os.environ['APPLICATION_ID'] = APP_ID
from google.appengine.api import apiproxy_stub_map,datastore_file_stub
from google.appengine.api.memcache import memcache_stub
apiproxy_stub_map.apiproxy = apiproxy_stub_map.APIProxyStubMap()
stub = datastore_file_stub.DatastoreFileStub( APP_ID, datastore_file=None )
apiproxy_stub_map.apiproxy.RegisterStub('datastore_v3', stub )
apiproxy_stub_map.apiproxy.RegisterStub('memcache', memcache_stub.MemcacheServiceStub())
class AppEngineFixture():
def __init__(self, request):
self.setup(request)
def setup(self, request):
self.test_env = testbed.Testbed()
self.test_env.activate()
self.test_env.init_datastore_v3_stub()
self.test_env.init_taskqueue_stub(True)
self.test_env.init_memcache_stub()
self.test_env.init_user_stub()
self.test_env.init_app_identity_stub(enable=True)
# self.test_env.init_all_stubs( enable=True)
request.addfinalizer(self.teardown)
def teardown(self):
self.test_env.deactivate()
@pytest.fixture
def mock_gae(request):
return AppEngineFixture(request)
def create_mock_request():
return Request.blank('')
@pytest.fixture
def mock_request():
return create_mock_request()
def create_mock_response():
return Response()
@pytest.fixture
def mock_response():
return create_mock_response()