Idea for a simple task queue

Edit, Feb 22, 2012

I have since updated and rewritten this project, which is described in this blog post. The source code is on github.

Original post

cue is a simple queue abstraction along the lines of the command pattern, and provides django apps with a way to decouple the creation of a request from its execution. cue is built on queues, a library which provides a uniform api to many popular storage backends (redis, sqs, beanstalkd, memcached, etc).

I'd like to point out here that if you are looking for the solution for implementing a task queue in python, celery is where it's at.

Show me your use-case

When a user performs an action on a site like Facebook, a notification needs to be propagated to all that users' friends and followers. If a user has 10 friends, it's not a big deal to create 10 records, but if the user has 1000 friends, it would significantly affect user experience to have to wait for the server to propagate all those records every time they want to do something. One solution is to enqueue the propagation, handling it in another process and allowing the user to go on their way.

How it works

Decoupling the requesting of an action from it's fulfillment. By using a simple message format, a command is enqueued and forgotten by the client, then later dequeued and executed. The command pattern provides a nice way of describing this interaction:

http://www.charlesleifer.com/photos/cue-class-diagram/

1. SomeView gets accessed, and an expensive or time-consuming operation now needs to be performed on SomeModel
2. In order not to bog down the request/response cycle, the operation will need to be performed out-of-process
3. Instantiate a command, ConcreteCommand, passing in SomeModel as its receiver
4. Tell the Invoker that the command needs to be executed
5. Some time later, the Invoker will get to the command, dequeueing it and calling its execute() method
6. The ConcreteCommand's execute() method encapsulates the logic needed to perform the expensive operations from step 1 on its receiver, SomeModel

What it looks like in code

The first thing is to define a command class that will execute the propagate() method on our receiver, which is assumed to be a model instance:

# commands.py
from cue.utils import QueueCommand

class PropagateActionCommand(QueueCommand):
    def execute(self):
        self.receiver.propagate()

    def undo(self):
        self.receiver.delete_propagated()

In some view, or maybe in a signal handler, you create the initial activity object that represents the action a user performed. To propagate the activity to a users' friends, a PropagateActivityCommand is instantiated with the action object and enqueued by the invoker:

from cue.utils import Invoker
from some_app.commands import PropagateActionCommand

invoker = Invoker()

def some_view(request):
    # ... create an action object
    action = Action.objects.create(...)
    command = PropagateActivityCommand(action)
    invoker.enqueue(command)

The last piece is the dequeue-ing and executing bit. This can be run in a daemon or on a cron, basically however you like. The code to execute enqueued commands is generic:

# in management command or daemon
invoker = Invoker()
invoker.dequeue() # dequeues a command and executes it

Try it out

Information on using cue in your project is available in the docs.

More information

• cue repo on GitHub
• cue docs
• celery: a task queue/job queue based on distributed message passing. It is focused on real-time operation, but supports scheduling as well.
• queues: a very simple API for interacting with lightweight queues
• command pattern

Comments (1)

• Andy | May 2010, at 05:27

  You mentioned celery is solution for implementing a task queue in python.

  How does cue differ from celery? What are the pros and cons?

  I'm all new to this, trying to figure out what would be a good choice for a beginner.

  ---

Commenting has been closed, but please feel free to contact me