Huey from Scratch: Step-by-Step Guide to

Understanding Huey: A Lightweight Task Queue for Python

Huey is a minimalist, yet powerful task queue library for Python designed to handle background jobs, scheduled tasks, and asynchronous execution without the overhead of heavier frameworks like Celery. Written by Charles Leifer, the creator of Peewee ORM, Huey prioritizes simplicity and ease of use while still delivering robust functionality for production workloads.

At its core, Huey allows you to define functions as tasks that get enqueued and executed by a separate worker process. This decoupling lets your web application or script return immediately while heavy lifting happens in the background—database updates, image processing, email sending, report generation, and more.

Why Huey Matters

In modern application development, blocking operations degrade user experience. A synchronous API call that triggers a 5-second image thumbnail generation forces the user to wait. Huey solves this by letting you offload work to a worker process. But why choose Huey over alternatives like Celery or RQ?

• Minimal footprint: Huey's codebase is small and auditable. You can read the entire source in an afternoon.
• No complex configuration: No brokers, no result backends to configure separately. Huey uses Redis, SQLite, or in-memory storage out of the box.
• Single-file worker: The worker can run as a standalone script or be embedded in your application.
• Task scheduling built-in: Cron-like scheduling, retries with exponential backoff, and task priorities come standard.
• Lightweight enough for small projects, robust enough for production: You don't need a distributed infrastructure to benefit from background tasks.

Installation and Setup

Install Huey along with a backend driver. For Redis support (recommended for production), install redis-py:

pip install huey redis

For SQLite-based persistence (great for single-server setups or development), the sqlite3 module is bundled with Python—no extra dependency needed. For in-memory storage (testing only), nothing additional is required.

Verify the installation:

python -c "import huey; print(huey.__version__)"

Defining Your First Task

Create a file named tasks.py. Import Huey, instantiate it with a backend, and decorate any function with @task() to register it as a Huey task:

from huey import Huey, RedisHuey

# Use Redis as the backend (recommended for production)
huey = RedisHuey(
    name='my-app',
    host='localhost',
    port=6379,
    db=0
)

# Alternatively, use SQLite for simple persistence
# from huey import SqliteHuey
# huey = SqliteHuey(name='my-app', filename='huey.db')

# Or in-memory for testing
# from huey import MemoryHuey
# huey = MemoryHuey()

@huey.task()
def send_welcome_email(user_id, email_address):
    print(f"Sending welcome email to user {user_id} at {email_address}")
    # Simulate some work
    import time
    time.sleep(2)
    print(f"Email sent to {email_address}")
    return f"Email delivered to {email_address}"

@huey.task()
def generate_report(report_id):
    print(f"Generating report #{report_id}...")
    time.sleep(3)
    print(f"Report #{report_id} generated successfully")
    return True

The @huey.task() decorator transforms a regular function into a task that, when called, doesn't execute immediately. Instead, it enqueues the function call and its arguments into the backend (Redis, SQLite, etc.), returning a special Result object that can later be checked for completion.

Enqueuing Tasks

From your application code—perhaps a Flask view or a script—import the decorated task functions and call them normally. The calls are non-blocking:

from tasks import send_welcome_email, generate_report

# These calls enqueue tasks immediately and return a Result object
result1 = send_welcome_email(42, 'user@example.com')
result2 = generate_report(101)

print(f"Task 1 enqueued. Result handle: {result1}")
print(f"Task 2 enqueued. Result handle: {result2}")

# You can optionally wait for results (blocking)
# final_result = result1.get(blocking=True, timeout=10)
# print(final_result)

The key insight: your main process returns instantly. The actual work happens in the worker process. If you call result.get(blocking=True), the caller waits until the worker finishes that specific task.

Running the Worker

The worker is a separate process that continuously polls the backend for new tasks and executes them. Start it from the command line, pointing to your tasks module:

huey_consumer tasks.huey

The argument tasks.huey follows Python module notation: it points to the huey instance inside tasks.py. The worker will output something like:

[INFO] Huey consumer starting with Redis backend
[INFO] RedisHuey: my-app
[INFO] Worker 1 started
[INFO] Processing task: send_welcome_email
[INFO] Task send_welcome_email completed in 2.01s

For production, run multiple workers to handle higher concurrency:

huey_consumer tasks.huey --workers 4

This spawns 4 worker threads (or processes with --worker-type process) that pull tasks from the queue in parallel.

Task Options and Configuration

Huey tasks accept several keyword arguments in the decorator to control behavior. Here are the most important ones:

@huey.task(
    priority=10,           # Higher number = higher priority (default: 0)
    retries=3,             # Number of retries on failure
    retry_delay=5,         # Seconds between retries
    context=True           # Pass task context as first argument
)
def process_payment(payment_id):
    # Task implementation
    pass

@huey.task(retries=3, retry_delay=10)
def call_external_api(endpoint, payload):
    # If this raises an exception, it will be retried up to 3 times
    response = requests.post(endpoint, json=payload)
    response.raise_for_status()
    return response.json()

When context=True is set, Huey passes a TaskContext object as the first argument to your function. This object contains metadata like the task ID, retry count, and the Huey instance itself, enabling dynamic behavior:

@huey.task(context=True)
def smart_task(ctx, data):
    if ctx.retries > 1:
        # Take a different code path after first retry
        print("This is retry attempt", ctx.retries)
    print(f"Processing data: {data} with task ID: {ctx.task_id}")
    # Access the Huey instance via ctx.huey
    return data

Scheduled Tasks and Periodic Jobs

Huey supports cron-like scheduling directly within the task decorator. Define tasks that run at fixed intervals or specific times:

from huey import crontab

@huey.task(crontab(crontab(minute='0', hour='*/2')))
def run_every_two_hours():
    print("This runs at minute 0 of every 2nd hour")
    # Perform periodic cleanup, sync, etc.

@huey.task(crontab(crontab(minute='30', hour='9', day_of_week='mon,fri')))
def monday_friday_report():
    print("Runs at 9:30 AM on Monday and Friday")

@huey.task(crontab(crontab(minute='*/15')))
def every_fifteen_minutes():
    print("Runs every 15 minutes")
    # Heartbeat check, cache refresh, etc.

The crontab utility supports standard cron expressions: minute, hour, day, month, day_of_week. Each can be a single value, a comma-separated list, a range, or a */N pattern.

Periodic tasks are enqueued by the scheduler, which runs inside the consumer process. No external scheduler (like Celery Beat) is required—the worker handles both regular tasks and scheduled tasks.

Retry Logic and Error Handling

By default, if a task raises an exception, Huey logs the error and marks the task as failed. With retries configured, Huey re-enqueues the task after retry_delay seconds, up to the specified limit. The retry count is tracked, so you can implement custom logic:

@huey.task(retries=5, retry_delay=10, context=True)
def resilient_task(ctx, url):
    import requests
    try:
        response = requests.get(url, timeout=5)
        response.raise_for_status()
        return response.text
    except requests.exceptions.RequestException as e:
        if ctx.retries < 3:
            # Retry normally
            raise  # Re-raise to trigger Huey's retry mechanism
        else:
            # After 3 attempts, fall back to a backup URL
            print("Falling back to backup endpoint")
            response = requests.get("https://backup.example.com/api")
            return response.text

You can also define a on_error callback to handle failures gracefully without blocking the worker:

def log_failure(task_id, exc):
    print(f"Task {task_id} failed with exception: {exc}")
    # Send to monitoring system, write to log, etc.

@huey.task(retries=2, on_error=log_failure)
def risky_operation():
    # If this fails, log_failure is called
    raise ValueError("Something went wrong")

Backends Deep Dive: Redis vs SQLite vs In-Memory

Huey abstracts storage behind a clean interface. Choosing the right backend depends on your deployment:

• RedisHuey: Best for multi-server, production environments. Redis provides fast, in-memory queue operations with persistence via RDB/AOF snapshots. Supports priorities natively. Requires a running Redis server.
• SqliteHuey: Ideal for single-server deployments, embedded applications, or development. No external dependency—tasks persist in a local .db file. Suitable for low-to-medium throughput. Not recommended for high-concurrency multi-process workers due to SQLite's locking model.
• MemoryHuey: Tasks live in process memory. Fastest for testing and prototyping but loses all tasks on process restart. Never use in production.

Backend configuration examples:

# Redis with custom connection pool settings
huey = RedisHuey(
    name='production-app',
    host='redis.internal',
    port=6379,
    password='secret',
    db=1,
    blocking_timeout=5,       # How long worker blocks waiting for tasks
    read_timeout=10,          # Redis socket read timeout
    max_connection_retries=3
)

# SQLite with WAL mode for better concurrency
huey = SqliteHuey(
    name='local-app',
    filename='/var/huey/tasks.db',
    cache_size=1024,          # SQLite page cache in KB
    journal_mode='wal'        # Write-Ahead Logging for concurrent readers
)

# In-memory with immediate execution (synchronous, for testing)
huey = MemoryHuey(immediate=True)

The immediate=True flag on any backend makes task calls execute synchronously—useful in unit tests where you want deterministic execution without running a worker.

Task Result Storage and Retrieval

By default, Huey stores task results in the backend for a configurable retention period. You can retrieve results using the Result object returned by task calls:

result = send_welcome_email(42, 'user@example.com')

# Check if the task is done (non-blocking)
if result.get(blocking=False) is None:
    print("Task not yet completed")

# Block until the task finishes (with timeout)
try:
    final_value = result.get(blocking=True, timeout=30)
    print(f"Task result: {final_value}")
except huey.exceptions.ResultTimeout:
    print("Task did not complete within 30 seconds")

# Access task metadata
print(f"Task ID: {result.task_id}")
print(f"Enqueued at: {result.enqueued_at}")

To prevent memory bloat in Redis or SQLite, configure result expiration:

huey = RedisHuey(
    name='my-app',
    result_ttl=3600  # Keep results for 1 hour (in seconds)
)

Task Pipelines and Chaining

Huey supports chaining tasks together so that the output of one task becomes the input of the next. Use the then() method on a Result object:

@huey.task()
def fetch_user_data(user_id):
    return {"id": user_id, "name": "John", "email": "john@example.com"}

@huey.task()
def enrich_user_data(user_data):
    user_data["enriched"] = True
    user_data["score"] = 95
    return user_data

@huey.task()
def store_user_data(enriched_data):
    print(f"Storing final data: {enriched_data}")
    # Save to database
    return True

# Chain them: fetch -> enrich -> store
result = fetch_user_data(42)
pipeline = result.then(enrich_user_data).then(store_user_data)

print(f"Pipeline started. Final result will be available via: {pipeline}")

Each then() call registers a follow-up task that receives the previous task's return value. The pipeline executes sequentially within the worker, ensuring data flows correctly through each stage.

Real-World Example: Async Image Processing

Consider a web application where users upload profile pictures. Generating thumbnails synchronously would block the request. Here's a complete Huey-based solution:

# image_tasks.py
from huey import RedisHuey
from PIL import Image
import os

huey = RedisHuey(name='image-processor', host='localhost', port=6379)

THUMBNAIL_SIZES = [(128, 128), (256, 256), (512, 512)]

@huey.task(retries=3, retry_delay=60)
def process_uploaded_image(filepath, user_id):
    """
    Task to generate thumbnails from an uploaded image.
    Retries up to 3 times with 60-second delays if something fails.
    """
    if not os.path.exists(filepath):
        raise FileNotFoundError(f"Uploaded file missing: {filepath}")
    
    original = Image.open(filepath)
    base_dir = f"/static/users/{user_id}/thumbnails"
    os.makedirs(base_dir, exist_ok=True)
    
    generated = []
    for width, height in THUMBNAIL_SIZES:
        thumb = original.copy()
        thumb.thumbnail((width, height))
        thumb_path = os.path.join(base_dir, f"{width}x{height}.jpg")
        thumb.save(thumb_path, "JPEG", quality=85)
        generated.append(thumb_path)
    
    # Clean up original upload to save space
    os.remove(filepath)
    
    return {
        "user_id": user_id,
        "thumbnails": generated,
        "count": len(generated)
    }

In your web view (Flask example):

# app.py
from flask import Flask, request, jsonify
from image_tasks import process_uploaded_image

app = Flask(__name__)

@app.route('/upload', methods=['POST'])
def upload_avatar():
    uploaded_file = request.files['image']
    user_id = request.form['user_id']
    
    # Save temporarily
    temp_path = f"/tmp/uploads/{uploaded_file.filename}"
    uploaded_file.save(temp_path)
    
    # Enqueue processing task (non-blocking!)
    result = process_uploaded_image(temp_path, int(user_id))
    
    # Return immediately with a task ID the client can poll
    return jsonify({
        "status": "processing",
        "task_id": result.task_id,
        "message": "Image uploaded. Thumbnails being generated."
    })

@app.route('/task/', methods=['GET'])
def check_task_status(task_id):
    # Retrieve result by task ID
    result = huey.result(task_id)
    if result is None:
        return jsonify({"status": "unknown_task"}), 404
    
    value = result.get(blocking=False)
    if value is None:
        return jsonify({"status": "pending"})
    elif isinstance(value, Exception):
        return jsonify({"status": "failed", "error": str(value)})
    else:
        return jsonify({"status": "completed", "data": value})

Graceful Shutdown and Worker Management

In production, you need to handle worker lifecycle gracefully. Huey's consumer responds to signals:

• SIGINT / SIGTERM: The worker finishes the current task and exits cleanly. No tasks are lost.
• SIGUSR1: (Linux/Unix) Toggles verbose logging for debugging.

To run Huey as a systemd service or Docker container, ensure the worker receives proper signals. Example systemd unit file:

[Unit]
Description=Huey Task Worker
After=network.target redis.service

[Service]
Type=simple
User=appuser
WorkingDirectory=/opt/myapp
ExecStart=/usr/bin/huey_consumer app.tasks.huey --workers 4
Restart=always
RestartSec=5

[Install]
WantedBy=multi-user.target

For Docker, run the consumer as the container's main process so it receives signals directly:

CMD ["huey_consumer", "app.tasks.huey", "--workers", "4"]

Best Practices for Huey in Production

• Keep tasks idempotent: Design tasks so they can be safely retried without duplicate side effects. Use transaction boundaries or unique keys to prevent double-processing.
• Set reasonable retry delays: Use exponential backoff for external API calls. A retry delay of 10 seconds with 3 retries gives services time to recover before giving up.
• Monitor queue depth: Periodically check the number of pending tasks in Redis (LLEN on the task list key) or SQLite table count. Alert when queues grow unexpectedly.
• Separate task definitions from worker startup: Keep your huey instance and task decorators in a module that both your application and the consumer import. Avoid circular imports.
• Use Redis for multi-worker setups: SQLite works for single-worker deployments but struggles with concurrent writers. Redis handles multiple workers naturally.
• Version your task signatures: When changing a task's parameters, consider backward compatibility or deploy workers before application code that enqueues the new signature.
• Handle result storage wisely: Set result_ttl to auto-expire old results. Storing millions of task results indefinitely will bloat your backend.
• Test with immediate=True: In unit tests, use MemoryHuey(immediate=True) to execute tasks synchronously without needing a running worker.
• Log task metadata: Use context=True to log task IDs, retry counts, and timing. This makes debugging failed tasks significantly easier.

Advanced: Custom Backends and Middleware

Huey's architecture is extensible. You can create custom backends by subclassing Huey and implementing storage methods. You can also hook into task lifecycle events via signals:

from huey import Huey, signals

huey = RedisHuey(name='my-app')

# Register signal handlers
@huey.signal(signals.TASK_STARTUP)
def on_task_start(task_id, task_name):
    print(f"Task {task_name} (ID: {task_id}) is starting")

@huey.signal(signals.TASK_COMPLETE)
def on_task_complete(task_id, task_name, result, duration):
    print(f"Task {task_name} completed in {duration:.2f}s. Result: {result}")

@huey.signal(signals.TASK_ERROR)
def on_task_error(task_id, task_name, exception, duration):
    print(f"Task {task_name} failed after {duration:.2f}s: {exception}")

Available signals include TASK_STARTUP, TASK_COMPLETE, TASK_ERROR, TASK_CANCELED, and TASK_LOCKED. These allow centralized logging, metrics collection, and alerting without modifying individual task functions.

Conclusion

Huey proves that a task queue doesn't need to be complex to be effective. With its clean API, built-in scheduling, retry support, and multiple backend options, it fits seamlessly into projects ranging from small scripts to production web services. By moving blocking operations into Huey tasks, you keep your application responsive while maintaining code clarity. The step-by-step patterns covered here—defining tasks, running workers, handling errors, chaining pipelines, and following production best practices—give you a complete foundation for integrating Huey into your Python ecosystem. Start with a single worker and a Redis backend, add scheduling as needed, and scale workers horizontally when traffic demands it. Huey's simplicity is its greatest strength: you'll spend less time configuring infrastructure and more time building features.