Python

on Python, App Engine

Safe deployment on App Engine

In a previous article I wrote about the need to update your Datastore indexes on App Engine before deploying code that relies on them. To simplify the deployment process, I put together a script that waits for your indexes to be built before deploying new code.

#!/usr/bin/env python

# Need to import and fix the sys path before importing other things.
import remote_api_shell
remote_api_shell.fix_sys_path()

import time

from google.appengine.api import datastore_admin
from google.appengine.ext.remote_api import remote_api_stub
from google.appengine.tools import appengine_rpc


APP_ID = 'your-app-id'


def configure_remote_api():
  def auth_func():
    return ('your.email@gmail.com', 'your.application.specific.password')

  remote_api_stub.ConfigureRemoteApi(
      APP_ID, '/_ah/remote_api', auth_func,
      servername='%s.appspot.com' % APP_ID,
      save_cookies=True, secure=False,
      rpc_server_factory=appengine_rpc.HttpRpcServer)
  remote_api_stub.MaybeInvokeAuthentication()


def main():
  print 'Checking indexes...'

  configure_remote_api()

  interval = 10  # seconds.
  building = True

  while building:
    # Start with a fresh run: maybe we're done building?
    building = False

    for index in datastore_admin.GetIndices('s~%s' % APP_ID):
      # If any single index is building, we're not done.
      # Sleep for a bit and then this cycle should repeat.
      if not index.has_state() or index.state() != index.READ_WRITE:
        building = True
        print 'Indexes are still building... Waiting %s seconds.' % interval
        time.sleep(interval)
        break

  print 'All indexes are up to date.'


if __name__ == '__main__':
  main()

Since this script will “block” until all indexes are built and ready, you can add this line to your deployment script, which should look something like...

$ appcfg.py update_indexes .
$ ./wait_for_indexes.py  # (The script above...)
$ appcfg.py update .
on Python

Python dictionary comprehensions

I’m a little embarrassed to say that I haven’t yet used these in my own code -- they seem so obvious now that I look back on it.

>>> d = {'a': 1, 'b': 2', 'c': 3}
>>> keys = ['a', 'c']
>>> { k: d[k] for k in keys }
{'a': 1, 'c': 3}
on Testing, ext.testbed, App Engine, Python

Task Queue support in App Engine's ext.Testbed

A while back I wrote some (now deprecated) code that allowed you to easily test your Python App Engine applications and their interaction with the App Engine APIs. When we got acquired by Google, I started working with some of the awesome App Engine engineers on making that code part of the official App Engine codebase.

We launched that, but one of the APIs that was noticeably lacking helper-methods was the Task Queue. I added one of the old methods (get_filtered_tasks()) but a bit later I noticed a pretty serious bug where timezones weren’t handled properly.

So that code hid quietly in the App Engine code base, undocumented and technically broken. Sorry :(

I wrote some extra code to fix it, but that didn’t get merged into the main repository for quite a while.

But now it’s here.

If you’re curious about the change, feel free to check out the diff (scroll to the bottom). The method is still called get_filtered_tasks() and takes several arguments as “filters” for various properties of tasks — and it should properly handle timezones when you set the eta or countdown properties when creating tasks.

Sorry for the delay. And enjoy.

PS: Since this isn’t documented on the official App Engine page, the API may change. I don’t intend to change it, but I can’t promise it won’t. Apologies in advance if you use this helper method and have it break out from under you.

Also, check out the official code.

Some example code:

import unittest
from google.appengine.api import taskqueue
from google.appengine.ext import testbed


class TaskQueueTestCase(unittest.TestCase):

  def setUp(self):
    self.testbed = testbed.Testbed()
    self.testbed.activate()
    self.testbed.init_taskqueue_stub()
    self.task_queue_stub = self.testbed.get_stub(testbed.TASKQUEUE_SERVICE_NAME)

  def tearDown(self):
    self.testbed.deactivate()

  def testTaskAdded(self):
    taskqueue.add(url='/path/to/task')

    tasks = self.taskqueue_stub.get_filtered_tasks(url='/path/to/task')
    self.assertEqual(1, len(tasks))
    self.assertEqual('/path/to/task', tasks[0].url)

unittest.main()