What is the simplest way to implement a remote FIFO queue as a Python GAE application?

Question

What is the simplest way to implement a remote FIFO queue as a Python GAE application and then push/pull name-value pair dictionaries to and from it?

For example, when an http get is made to the GAE application, the GAE app would return the oldest collection of name-value pairs that were posted to the app which have not been previously pulled from the queue. These name-value pairs would then be re-instantiated as a dictionary on the client side. urllib.urlencode provides a simple mechanism to encode dictionaries as parameters, but what is the similarly simple approach to decode parameters into dictionaries when you http "get" them? When there are no items in the queue, the GAE app should return a null or some other more appropriate identifier the client could respond to.

#A local python script
import urllib 
targetURL="http://myapp.appspot.com/queue"

#Push to dictionary to GAE queue
params = urllib.urlencode({'spam': 1, 'eggs': 2, 'bacon': 0})
f = urllib.urlopen(targetURL, params)
print f.read()
params = urllib.urlencode({'foo': 1, 'bar': 2})
f = urllib.urlopen(targetURL, params)
print f.read()


#Pull oldest set of name-value pairs from the GAE queue and create a local dictionary from them.
#f = urllib.urlopen(targetURL, ……)
#returnedDictionary = ????

What would the simplest way to implement this short GAE application?

#queue.py a url handler in a GAE application.  
# For posts, create an object from the posted name-value pairs and insert it into the queue as the newest item in the queue
# For gets, return the name-value pairs for the oldest object in the queue and remove the object from the queue.
#   If there are no items in the queue, return null

Answer 1

The below assumes you're using the webapp framework.

The simple answer is that you just use self.request.GET, which is a MultiDict (which you can treat as a dict in many cases) containing the form data sent to the request.

Note that HTTP allows form data to contain the same key multiple times; if what you want isn't really a dict but a list of key-value pairs that have been sent to your application, you can get such a list with self.request.GET.items() (see http://pythonpaste.org/webob/reference.html#query-post-variables ) and then add these pairs to your queue.

Answer 2

Something along these lines:

from google.appengine.ext import db
from google.appengine.ext import webapp
from google.appengine.ext.webapp import run_wsgi_app

class QueueItem(db.Model):
  created = db.DateTimeProperty(required=True, auto_now_add=True)
  data = db.BlobProperty(required=True)

  @staticmethod
  def push(data):
    """Add a new queue item."""
    return QueueItem(data=data).put()

  @staticmethod
  def pop():
    """Pop the oldest item off the queue."""
    def _tx_pop(candidate_key):
      # Try and grab the candidate key for ourselves. This will fail if
      # another task beat us to it.
      task = QueueItem.get(candidate_key)
      if task:
        task.delete()
      return task
    # Grab some tasks and try getting them until we find one that hasn't been
    # taken by someone else ahead of us
    while True:
      candidate_keys = QueueItem.all(keys_only=True).order('created').fetch(10)
      if not candidate_keys:
        # No tasks in queue
        return None
      for candidate_key in candidate_keys:
        task = db.run_in_transaction(_tx_pop, candidate_key)
        if task:
          return task

class QueueHandler(webapp.RequestHandler):
  def get(self):
    """Pop a request off the queue and return it."""
    self.response.headers['Content-Type'] = 'application/x-www-form-urlencoded'
    task = QueueItem.pop()
    if not task:
      self.error(404)
    else:
      self.response.out.write(task.data)

  def post(self):
    """Add a request to the queue."""
    QueueItem.push(self.request.body)

One caveat: Because queue ordering relies on the timestamp, it's possible for tasks that arrive very close together on separate machines to be enqueued out-of-order, since there's no global clock (only NFS synced servers). Probably not a real problem, depending on your use-case, though.