Mini-Huey

MiniHuey provides a very lightweight huey-like API that may be useful for certain applications. The MiniHuey consumer runs inside a greenlet in your main application process. This means there is no separate consumer process to manage, nor is there any persistence for the enqueued/scheduled tasks; whenever a task is enqueued or is scheduled to run, a new greenlet is spawned to execute the task.

MiniHuey may be useful if:

  • Your application is a WSGI application.

  • Your tasks do stuff like check for spam, send email, make requests to web-based APIs, query a database server.

  • You do not need automatic retries, persistence for your message queue, dynamic task revocation.

  • You wish to keep things nice and simple and don’t want the overhead of additional process(es) to manage.

MiniHuey may be a bad choice if:

  • Your application is incompatible with gevent (e.g. uses asyncio).

  • Your tasks do stuff like process large files, crunch numbers, parse large XML or JSON documents, or other CPU or disk-intensive work.

  • You need a persistent store for messages and results, so the consumer can be restarted without losing any unprocessed messages.

If you are not sure, then you should probably not use MiniHuey. Use the regular Huey instead.

Usage and task declaration:

class MiniHuey([name='huey'[, interval=1[, pool_size=None]]])
Parameters:

  • name (str) – Name given to this huey instance.

  • interval (int) – How frequently to check for scheduled tasks (seconds).

  • pool_size (int) – Limit number of concurrent tasks to given size.

task([validate_func=None])

Task decorator similar to Huey.task() or Huey.periodic_task(). For tasks that should be scheduled automatically at regular intervals, simply provide a suitable crontab() definition.

The decorated task will gain a schedule() method which can be used like the TaskWrapper.schedule() method.

Examples task declarations:

from huey import crontab
from huey.contrib.mini import MiniHuey

huey = MiniHuey()

@huey.task()
def fetch_url(url):
    return urlopen(url).read()

@huey.task(crontab(minute='0', hour='4'))
def run_backup():
    pass

Example usage. Running tasks and getting results work about the same as regular Huey:

# Executes the task asynchronously in a new greenlet.
result = fetch_url('https://google.com/')

# Wait for the task to finish.
html = result.get()

Scheduling a task for execution:

# Fetch in ~30s.
result = fetch_url.schedule(('https://google.com',), delay=30)

# Wait until result is ready, takes ~30s.
html = result.get()
start()

Start the scheduler in a new green thread. The scheduler is needed if you plan to schedule tasks for execution using the schedule() method, or if you want to run periodic tasks.

Typically this method should be called when your application starts up. For example, a WSGI application might do something like:

# Always apply gevent monkey-patch before anything else!
from gevent import monkey; monkey.patch_all()

from my_app import app  # flask/bottle/whatever WSGI app.
from my_app import mini_huey

# Start the scheduler. Returns immediately.
mini_huey.start()

# Run the WSGI server.
from gevent.pywsgi import WSGIServer
WSGIServer(('127.0.0.1', 8000), app).serve_forever()
stop()

Stop the scheduler.

Note

There is not a separate decorator for periodic, or crontab, tasks. Just use MiniHuey.task() and pass in a validation function. A validation function can be generated using the crontab() function.

Note

Tasks enqueued for immediate execution will be run regardless of whether the scheduler is running. You only need to start the scheduler if you plan to schedule tasks in the future or run periodic tasks.