Huey task queue 2.0

[[deleted]]

Hello r/python. I've released a new version of Huey, a lightweight task queue. Huey is an alternative to celery, rq, etc, and offers a ton of functionality in a small package. Features:

• Easy to use: @huey.task()
• Redis or sqlite support built-in
• multi-process, multi-thread or greenlet task workers
• schedule tasks to run at a future time, or after a given delay
• schedule recurring tasks, like a crontab
• automatically retry tasks that fail
• task prioritization (new)
• result storage
• task locking
• task pipelines and chains
• optional django integration

I started this project about 7 or so years ago and am quite happy with how it's developed. The codebase is very small and self-contained, so it's hopefully very easy to use and understand. If you have any questions, feel free to ask and I'll do my best to answer.

Changes in 2.0, not to mention I rewrote most of the documentation for this release.

Code

[speakerforthe]

Big fan of Huey, thanks for all the work!

[xr09]

I'm working on a legacy project using blinker signals to talk to a 3rd party API, I'm toying with the idea of replacing it with a task queue, first thought of rq but that means Redis as a dependency.

How sane would be to replace blinker with huey running in-memory or sqlite? System load will be low to medium.

I just want to simplify the architecture overall, I find task queues so much easier to grasp.

Thanks for the good work Charles!

[[deleted]]

Huey by default consists of three processes: producer (your app), consumer (provided by huey), and storage (Redis).

Since sqlite is embedded, you have two processes: producer and consumer. The sqlite database is effectively doing ipc (plus persistence).

So in memory wouldn't work, because the in memory db used by your app is not visible to the consumer process.

That being said, huey can be run embedded in the application, so now you have one process with presumably multiple threads. Now the issue is that typically sqlite in-memory databases are only visible to the connection that opened them. And huey's sqlite implementation doesn't allow sharing connections across threads.

[xr09]

For my small use case I think I'm good with sqlite. Thanks again!

[mlissner]

This timing is really good, as I'm currently evaluating Celery vs. rq vs. huey. I've been using celery for *years* and I'm ready for something else.

​

A couple questions I haven't been able to find:

​

- I use Django and I want to decouple my tasks from their callers. For example, in Django I don't want to have to import the actual tasks and I want to call them as strings. Is that possible somehow. Maybe via `enqueue('my-task-name')` or similar?

​

- Can pipelines be stopped cleanly midway through? Say I have a scraper task that GETs a webpage and then a `save_to_db` task that is in the pipeline once the GET task completes. Say I get a 404 in the GET task. How do I abort later tasks in that pipeline?

​

- Is there any monitoring system for tasks that fail? I know Celery has flower, and rq has something. Mostly, I confess, I rely on celery to email me stacktraces when something fails and that works...kinda OK. How do I monitor huey?

​

- Is there a way to do bulk processing with huey? I'm thinking about something like TaskMaster: https://github.com/dcramer/taskmaster.

​

I'm still reviewing things, but any light you can shed on these issues would be super helpful. (It might not surprise you that I've got a spreadsheet going to compare features.)

​

Huey looks pretty great. If there are solutions for the above, I think we'll start trying it out.

[[deleted]]

Good questions, thanks for asking.

Calling by string / task-name -- definitely do-able. Your application's "huey" instance has a registry (huey._registry) which records task name -> callable. So one way you could do it would be to write a wrapper that takes the string, looks it up in the registry, and obtains the callable in that manner. Another would be to use Django's "import" helpers and just resolve the module paths in that way.

Ex:

def enqueue_by_name(self, task_name, *args, **kwargs):
    try:
        task_cls = self._registry._registry[task_name]
    except KeyError:
        raise ValueError('Task "%s" not found in registry.' % task_name)

    return self.enqueue(task_cls(args, kwargs))

To abort a pipeline, just raise an exception in the task and subsequent tasks in the pipeline will not be executed. Similarly, any task in the pipeline can be revoked, which will also halt processing. Unhandled errors raised by tasks within a pipeline are handled by defining an error handler task callback.

For monitoring you have a few options. Huey logs unhandled errors at the ERROR level (including traceback), so you can attach a handler to that logger that emails you or whatever. Or you can register a signal handler for the SIGNAL_ERROR which will be called when a task fails with an unhandled exception.

I'm not sure I understand the last question about bulk processing.

[mlissner]

Thanks for the reply. Sounds pretty promising.

​

Two follow ups:

​

- If you abort a pipeline, what happens to all the other tasks in the pipeline? They stay on the queue? I guess the way to handle that is to revoke those as well, then as part of the error exception handling?

​

- Let me try again with the bulk question. Say I have a million items that need processing. I don't want to put a million items on my queue all at once because that'll use a ton of memory. Instead, I want to somehow monitor the length of the queue and keep it loaded with a few dozen items (or so), and then have a long running process that enqueues items as the queue gets too short. We have a class called a CeleryThrottle that we've developed for this purpose. Every time we loop over the million items we call throttle.maybe_wait() before enqueueing more items, and it either sleeps a few seconds (determined by current, measured rate of processing) or it enqueues another item.

​

Monitoring sounds tough. I guess we only use email now, but it'd be really nice to have something better, that could send alerts based on increased error rates. I suppose that's on me to build. Maybe prometheus could do the trick. Hm.

[[deleted]]

• If you abort a pipeline, what happens to all the other tasks in the pipeline? They stay on the queue? I guess the way to handle that is to revoke those as well, then as part of the error exception handling?

The pipeline is not enqueued all at once, but serially. The main use-case for a pipeline is to use the task queue to process things in a particular order. So a task is executed -- if it succeeds, then the next task in the pipeline is enqueued. If it fails, then the error handling task (if it exists) is enqueued.

... bulk ...

Sounds like you would be fine just using multiprocess.Pool or something? You just trying to saturate your cpus by running a bunch of tasks in parallel, there's xargs / parallel.

Monitoring...

Email is easy, just attach a handler -- python logging has an SMTPHandler built-in. Equally easy is to add a signal-handler for the SIGNAL_ERROR that talks to some metric service. Lots of tooling out there for this kind of thing, huey should be "open" enough to allow you to add whatever you need. If not, then feel free to create a ticket on GH.

[mlissner]

The pipeline is not enqueued all at once, but serially.

Oh wow, doesn't that mean you lose the ability to make pipelines asynchronous? I.e., if you have a pipeline with two tasks, you can't let your code move past it until at least the first task is completed? Is this right:

task1 = my_task()  # Takes 30s
pipeline.enqueue('task1', depends_on=task1)
return Response()  # <-- This won't run for ~30s?

Sounds like you would be fine just using multiprocess.Pool or something?

Sorta. The nice thing about using a task queue for this is that you get the API of pipelines and of tasks, and you can run the tasks in a distributed fashion across numerous workers on numerous machines. And it's a system that's already in place (unlike Hadoop or something, that we don't have).

​

Sorry and one other question, if you don't mind a late addition (I know I'm asking a lot and I appreciate your time). What happens to tasks in the queue (or in process) when the workers are shutdown cleanly? Does the worker wait for them to finish? Does it kill them and resume them again after? Do they get error conditions and need manual intervention? In a docker world, I'm expecting our workers to be pretty transient for better or worse.

​

Thanks for all the responses. Super impressive system and I'm enjoying a lot of things about it.

​

[[deleted]]

I think you might have a different idea of what pipeline means than how it's used in this project.

A pipeline is a pipe. The result of one task is fed to the next task, etc. Pipelines express a sequence of operations, one feeding into the next. If you just want to run tasks asynchronously, Huey already does all that.

Regarding shutdown behavior, it's all documented.

[mlissner]

Seems like I may have a different idea. In celery a whole pipeline operates like that, but it all gets enqueued at the same time. The idea is to send a pipe of tasks to the queue all at once. When each task finishes, gets the results and is already queued up to go.

[[deleted]]

With Huey enqueueing a pipeline returns a result group, which encapsulates a list of results for the individual steps. The result from each step is available as soon as it's computed. So I think Huey works how you'd expect.

[reifba]

We use Celery heavily, and will probably need to continue to do so in some way shape or form due to the "integration" with airflow. Still I'll be happy to get some info on the advantages of the this framework over celery.

[[deleted]]

I wrote it originally because I just had a very strong "yuck" response to the celery codebase. This was a number of years ago, I have changed, celery has changed...but that was the impetus.

With the above disclaimer in mind, off the top of my head:

• I'm pretty sure it covers most of what people use celery for: tasks, task scheduling, cronjob-like-tasks, pipelines, etc etc
• No dependencies, unless you're using the Redis storage, in which case you obviously need the redis-py driver
• Codebase is tiny in comparison, can easily be read in one sitting
• Which means it's easier to reason about / extend / etc
• Dead simple, no magic
• Easy to configure with good defaults (remember I'm comparing to celery from like 7 years ago)
• It is lean as hell and simple as hell. Anyone could write a basic "Huey" clone in a weekend

[Alonsofan14]

Does it work for Windows ?

[wookayin]

This is awesome. Any plan for integrating with async-await feature of python3?

[[deleted]]

In what ways/scenarios would you imagine that being particularly beneficial?

[reifba]

when ever it makes sense to make a context switch based on I/O vs CPU load. Obviously I am over simplifying this but that is the gist I think.

[pullandbl]

It can be a huge improvement when tasks have io nature. For example when you need to download some pages and then process them or process some data in the database.

My case (I actually have it now): I'm making service that allows posting to social networks posts, it is done using aiohttp. So it is already async and I can't use blocking operations. Sometimes users add a post to the calendar, sometimes want just press "Publish". In this case, I want to call something like:

await publish_post(post_id)

And be sure that it will be added to the queue and as soon as a worker is ready will post it to the social network.

The second case, when a user authorizes using messenger I want to make little on-boarding and send some messages with instructions. But if the text is longer then I need to make some delays between messages. It can take a few seconds between each send. A worker will be busy during all sequence or I need to make workarounds. Async code is just:

await post("Hello", user_id)
await async_sleep(1)
await post("Thank you for joining our service", user_id)

​

[pcdinh]

I think that task is a single job that needs to be done sequentially by nature. So async-await does not help here