An google-appengine loading module to aid writing scripts for GAE

ae.py

######################################################
# allow scripts to use appengine apis
#
# assumes you store your data in a "data" directory 
# in your app directory. See connect_local_datastore()
# and assumes that you have a HRD app id with "~"
######################################################
#
# locate app-engine SDK:
AE_PATH = "/your/path/to/sdk/google_appengine/"
#
# Your username:
AE_USER = '[email protected]'
######################################################

import getpass
import sys
import os

# fix timezone
import time
os.environ['TZ'] = time.tzname[1]
import datetime

# load the AE paths (as stolen from dev_appserver.py)
APP_PATH = os.path.abspath( os.path.join(os.path.dirname(__file__), '..') )
EXTRA_PATHS = [
    APP_PATH,
    AE_PATH,
    os.path.join(AE_PATH, 'lib', 'antlr3'),
    os.path.join(AE_PATH, 'lib', 'django'),
    os.path.join(AE_PATH, 'lib', 'django_1_2'),
    os.path.join(AE_PATH, 'lib', 'ipaddr'),
    os.path.join(AE_PATH, 'lib', 'protorpc'),
    os.path.join(AE_PATH, 'lib', 'webob'),
    os.path.join(AE_PATH, 'lib', 'yaml', 'lib'),
    os.path.join(AE_PATH, 'lib', 'fancy_urllib'), #http://code.google.com/p/google-app-engine-django/issues/detail?id=178
    os.path.join(AE_PATH, 'lib', 'graphy')
]
sys.path = EXTRA_PATHS + sys.path

# shhhhhhh
import warnings
warnings.simplefilter('ignore', DeprecationWarning)

# choose django
from google.appengine.dist import use_library
use_library('django', '1.2')

# work out what our app-id is
cmd = "(cd %s && grep 'application:' app.yaml | cut -c 13-)" % APP_PATH
APP_ID = os.popen(cmd).read().strip()
if not APP_ID:
    raise Exception("Could not work out app id")
os.environ['APPLICATION_ID'] = APP_ID

import logging

def connect_remote_datastore():
    """Run this from scripts to get access to the PRODUCTION devserver"""
    import getpass
    import os
    import threading
    from google.appengine.ext.remote_api import remote_api_stub
    from google.appengine.ext import db
    from google.appengine.ext import blobstore
    from google.appengine.ext.remote_api import throttle
    passw = None
    def auth_func():
        passw = os.environ.get('GAE_PASS')
        if not passw:
            passw = getpass.getpass("password:")
            os.environ['GAE_PASS'] = passw
        return AE_USER,passw
    host = '%s.appspot.com' % APP_ID
    ns = os.environ.get('GAE_NAMESPACE','')
    remote_api_stub.ConfigureRemoteDatastore(
        "s~%s" % APP_ID, 
        '/remote_api', 
        auth_func, 
        host, 
        save_cookies=True, 
        secure=True)
    # full_throttle = throttle.DefaultThrottle(multiplier=10.0)
    # throttle.ThrottleRemoteDatastore(full_throttle)
    # full_throttle.Register(threading.currentThread())
    logging.info("connected to %s datastore using %s namespace" % (host,ns))
    # set ns
    from google.appengine.api import namespace_manager
    namespace_manager.set_namespace( ns )
        
def connect_local_datastore():
    """Run this from scripts to get access to the local devserver"""
    from google.appengine.tools import dev_appserver
    from google.appengine.tools.dev_appserver_main import ParseArguments
    argv = [
        sys.argv[0],
        '--datastore_path', os.path.join(APP_PATH,'data','datastore.db'),
        '--rdbms_sqlite_path', os.path.join(APP_PATH,'data','datastore.sqlite'),
        '--blobstore_path', os.path.join(APP_PATH,'data','blobstore'),
        '--history_path', os.path.join(APP_PATH,'data','history'),
        '--use_sqlite',
        '--require_indexes',
        '--skip_sdk_update_check'
    ]
    args,opts = ParseArguments(argv) # TODO allow opts
    dev_appserver.SetupStubs(os.environ['APPLICATION_ID'], **opts)
    # set ns
    from google.appengine.api import namespace_manager
    ns = os.environ.get('GAE_NAMESPACE','')
    logging.info("connected to local datastore using %s namespace" % ns)
    namespace_manager.set_namespace( ns )