Skip to main content

Chaining tasks on Google Appengine's (GAE) taskqueue.

Project description

A convenience module on top of the deferred library that comes with the Google AppEngine (GAE).

In a nutshell:

from waterf import queue, task

        task(remove, id=101),
        task(remove, id=102),
        task(remove, id=103)
    task(email, to='')

Should be pretty self-explanatory: it first runs the function check_condition, then it runs the function remove three times in parallel, after that it runs email.

To abort execution of a series you either raise queue.PermanentTaskFailure or as a convenience return queue.ABORT. If you return another task, you further defer so to speak: the original task will get resolved (or aborted) as soon as the new (returned) task gets resolved (or aborted).

You use task() exactly the same as you used deferred.defer():

task(check, id=102, _countdown=20)
task(email, to='', _queue='mailer')

After constructing a task you enqueue() it; the relation to the deferred.defer is roughly speaking:

task(foo, 'bar').enqueue()  <==> deferred.defer(foo, 'bar')
task(foo, 'bar').run()      <==> foo('bar')

Enqueue’ing takes (again) the same options defer took, overruling the ones you used in the constructor, e.g.:

task(foo).enqueue(queue='mailer', countdown=60)

waterf adds two options:

use_id  True | False | str
        Use if you don't come up with a good name to prevent double-scheduling
        The value True means autogenerate a good id, otherwise takes your str
        Defaults to True if a name is not set, otherwise to False

release_after <seconds>
        Determines when the id will be released after your task has finished
        Defaults to 0, immediately

Tasks implement a jquery-like callback interface:

task(foo).then(email_user, email_admin).always(...)

The callbacks must accept as their first argument the message the task sent. But this message passing will likely be dropped in a future version, because it’s unused by the library.

On top of the waterf.queue there is some experimental jet set in the waterf.snake module, which implements a ndb.tasklet like api:

from waterf import snake

def A():
    raise snake.Return('A')

def B(): ...

def work():
    anA = yield snake.task(A)
    yield snake.task(B), snake.task(C) ...  # parallel yield


Note that you have to enable the deferred library in your app.yaml

- deferred: on

Thank you.

Project details

Release history Release notifications

This version
History Node


History Node


History Node


History Node


History Node


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Filename, size & hash SHA256 hash help File type Python version Upload date (16.8 kB) Copy SHA256 hash SHA256 Source None Jun 3, 2013

Supported by

Elastic Elastic Search Pingdom Pingdom Monitoring Google Google BigQuery Sentry Sentry Error logging CloudAMQP CloudAMQP RabbitMQ AWS AWS Cloud computing Fastly Fastly CDN DigiCert DigiCert EV certificate StatusPage StatusPage Status page