john pfeiffer

google app engine boto s3 import error boto.pyami.config

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()

Published

Category

python-appengine

~412 words

Tags