Handlers

[jeanphix]

• Make expected_at part of Task payload
• Add few helpers to make task callbacks management more friendly:

from pq.handlers import (
    handler,
    Worker,
)

@handler(queue, schedule_at='1h')
def eat(kind):
    print 'umm, %s apples taste good.' % kind

eat('Cox')

Worker(queue).work()
# One hour later:
# > umm, Cox apples taste good.

[stas]

@jeanphix while I can see the benefit of the pq.handlers module you've added, I find it very optional.

To give you an idea, we use pq with thread-based workers that have a lot of extra features like logging, self-healing, custom statuses, etc. We also have worker manager to manage the workers life-cycle and etc... The beauty of the pq is that it is very simple and flexible. Now introducing a worker like the one provided by you makes it opinionated in some way, similar to celery.

I think it is great that you decided to contribute back your work on pq, but my opinion is that this should not be part of the core library in order to keep things simple. It might be a better idea to start a project like pq-contrib for this kind of extra functionalities.

@malthe it would be great to hear your opinion on this too.

[stas]

@jeanphix the other part with extra Task fields looks great though. Thank you!

[malthe]

I need to look this through in-depth. I don't really mind having some framework-like functionality but it's true a small library is often a good library.

[jeanphix]

@stas @malthe The thing is that it's quite a redundant work having to switch on task payload to fire the appropiate stuff within the codebase.

I designed this extension (within a dedicated submodule) to make it more easy to handle. It's also an entry point to abstract other stuff like rescheduling. It could be quite cool to have something like:

@handler(queue, schedule_at='1h', max_retries=3, retry_at='10s')
my_task(param)
    return param

my_task('my value')

Worker(queue).work()

that just works out of the box...

Not sure it makes sense to create an extra repos for such optional high level features cause it will then be more difficult to keep in sync.

May just be moved to ext.handlers or contrib.handlers?

[malthe]

@jeanphix – I think it looks pretty good actually.

Why is it called handler though? How about:

@task(queue, schedule_at='1h', max_retries=3, retry_at='10s')
def foo(param):
    return param

# Just call directly.
foo('my value')

# Schedule for later processing.
foo.delay()

Worker(queue).work()

But then it's kind of like having a Celery backend. I wonder if it's hard to make pq work as a Celery backend instead.

[jeanphix]

@malthe yes, task could be a better naming, but it will then kind of clash with the core Task class.

In other frameworks it's used to be Job vs Task where Job is a single callback of a given Task.

Celery is an heavy library... That's why people will choose pq :)

So, what about:

• renaming pq.Task as pq.Job
• renaming pq.handlers.handler as pq.helpers.task

[malthe]

I think I agree with that.

A task is like a class where a job is a specific instance of that task. It might be nice to allow the form @queue.task(...) too.

[jeanphix]

Ok, will fix it.

[jeanphix]

@malthe So basically:

• Task has been renamed into Job
• The new tasks API now looks like:

from pq.tasks import PQ

pq = PQ()

queue = pq['default']

@queue.task(schedule_at='1h')
def my_task(argument);
    return argument

my_task('value')

queue.work()

The core API is not poluate which should make @stas happy.

[stas]

As I said, I'm still not convinced this should be part of the core library, but I trust @malthe in his intentions.

Still, if we would be serious about developing a framework around the concept of worker for pq, here's a list of things I personally miss in pq and had to work around that:

• pluggable back-end API, probably safest is to have a class-based worker API similar to the one provided by the multiprocessing and threading module.
• logging support, provide persistence not just to answer the questions of when? and what? but also how? a task/job was executed
• task/job status support, with the current API you can not tell which task failed and which didn't, a worker should mark a task as successful/failed/in-progress at least
• workers management, provide a tool to help manage (start/stop) and scale (basic support for amount of workers to be spawned)

These are just some basic questions I have in mind when I think about a queue. In fact I will be happy to share some of the work we did using pq. Just let me know if this sounds interesting, I was not feeling confident about sharing it until this question popped. :)

[jeanphix]

workers management, provide a tool to help manage (start/stop) and scale (basic support for amount of workers to be spawned)

The host system is responsible of this, not the app itself.

[jeanphix]

@malthe ^ works as expected, c.f.: https://github.com/jeanphix/pq/tree/handlers#tasks

[malthe]

Nice work. I think this is just about ready to merge.

[jeanphix]

@malthe Possible to get a release? Cheers!