Idea for a simple task queue

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:

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. The docs provide an example of queueing comments for processing by a spam filter.

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)

Commenting has been disabled for this entry