Huey for Python Background Jobs
In this tutorial, you will learn about Huey for Python Background Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Huey is a lightweight Redis-backed task queue for Python with scheduling, retries, priorities, and a simple API for background job processing.
What You Learn
You will learn how to set up Huey, define tasks with decorators, configure task scheduling, handle retries, use the result backend, and run the Huey consumer.
Why It Matters
Huey is simpler than Celery with zero configuration needed for basic usage. It is ideal for smaller projects, scripts, and applications that need background job processing without the complexity of a full task queue system.
Real-World Use
DodaTech uses Huey for lightweight background tasks: cache warming, log rotation, health checks, and simple email notifications. For complex workflows, they use Celery.
Basic Setup
from huey import RedisHuey
import time
huey = RedisHuey('my_app', host='localhost', port=6379)
@huey.task()
def send_email(to, subject):
print(f"Sending email to {to}: {subject}")
time.sleep(1)
print(f"Email sent to {to}")
return True
# Enqueue
result = send_email('user@example.com', 'Welcome')
print(f"Task ID: {result.id}")
Task Decorators
from huey import RedisHuey
import time
huey = RedisHuey('tasks')
# Simple task
@huey.task()
def process_data(data):
return f"Processed: {data}"
# Task with retries
@huey.task(retries=3, retry_delay=10)
def fetch_url(url):
import requests
response = requests.get(url, timeout=5)
return response.status_code
# Task with priority
@huey.task(priority=10)
def urgent_task(data):
return f"Urgent: {data}"
# Periodic task (every 60 seconds)
@huey.periodic_task(crontab(minute='*/1'))
def heartbeat():
print(f"Heartbeat at {time.time()}")
Task with Context
from huey import RedisHuey, crontab
import time
huey = RedisHuey('context_demo')
@huey.task(context=True)
def process_image_task(task, image_path):
print(f"Processing image: {image_path}")
print(f" Task ID: {task.id}")
print(f" Retry count: {task.retries}")
for i in range(5):
time.sleep(0.5)
print(f" Progress: {(i+1)*20}%")
return {'path': image_path, 'status': 'done'}
result = process_image_task('/uploads/photo.jpg')
print(f"Result: {result.get(blocking=True, timeout=30)}")
Scheduling
from huey import RedisHuey, crontab
import time
huey = RedisHuey('scheduler')
# Run every 30 minutes
@huey.periodic_task(crontab(minute='*/30'))
def clean_temp_files():
print("Cleaning temporary files...")
time.sleep(1)
print("Cleanup complete")
# Run daily at 3 AM
@huey.periodic_task(crontab(hour='3', minute='0'))
def daily_report():
print("Generating daily report...")
time.sleep(5)
print("Report generated")
# Run every 5 seconds (for testing)
@huey.periodic_task(crontab(minute='*', second='*/5'))
def quick_check():
print(f"Health check at {time.strftime('%H:%M:%S')}")
Result Storage
from huey import RedisHuey
import time
huey = RedisHuey('results')
@huey.task()
def compute_pi(iterations):
pi = 0
for i in range(iterations):
pi += ((-1) ** i) / (2 * i + 1)
return pi * 4
@huey.task()
def multiply(x, y):
return x * y
# Execute and check results
result = compute_pi(1000000)
print(f"Task queued: {result.id}")
# Block until complete
pi_value = result.get(blocking=True, timeout=30)
print(f"Pi: {pi_value:.10f}")
# Check without blocking
r2 = multiply(6, 7)
time.sleep(0.5)
print(f"Ready: {r2()}") # Returns result or None
if r2():
print(f"6 * 7 = {r2()}")
Task Priority
from huey import RedisHuey
import time
huey = RedisHuey('priority_queue')
@huey.task(priority=10)
def critical(msg):
print(f"[CRITICAL] {msg}")
return "critical"
@huey.task(priority=5)
def normal(msg):
print(f"[NORMAL] {msg}")
return "normal"
@huey.task(priority=1)
def background(msg):
print(f"[BACKGROUND] {msg}")
return "background"
# Tasks with higher priority are processed first
normal("user action")
background("cleanup task")
critical("system alert")
Running the Consumer
# Start Huey consumer (single thread)
huey_consumer.py my_app.huey
# With multiple worker threads
huey_consumer.py my_app.huey --workers 4
# With log level
huey_consumer.py my_app.huey --workers 4 --logfile huey.log --verbose
# my_app.py
from huey import RedisHuey
huey = RedisHuey('my_app')
@huey.task()
def process(item):
return f"Processed: {item}"
Common Mistakes
1. Forgetting to Run the Consumer
Huey tasks do not execute without a running consumer. Always start huey_consumer.py in a separate Process.
2. Using Blocking .get() Without Timeout
result.get(blocking=True) blocks indefinitely. Always set a timeout to prevent worker hangs.
3. Not Setting retries for Unreliable Operations
Network calls and external API requests should have retries. Set retries=3 and retry_delay=10 for these tasks.
4. Running Too Many Workers
Huey workers use threads. Too many threads cause GIL contention. Start with 2-4 workers, adjust based on CPU usage.
5. Missing crontab Import
from huey import crontab is required for periodic tasks. Forgetting the import causes NameError.
Practice Questions
1. How does Huey differ from Celery?
Huey is simpler with zero configuration for basic tasks. Celery has more features (complex routing, multiple brokers, result backends, monitoring). Huey is better for simpler applications.
2. How do you define a periodic task in Huey?
Use the @huey.periodic_task(crontab(...)) decorator. The crontab expression defines the schedule.
3. How do you get a task result in Huey?
Call result.get(blocking=True, timeout=N). Or check with result() which returns None if not ready.
4. How do you start the Huey consumer?
Run huey_consumer.py path.to.huey_instance. Use --workers N for concurrency.
Challenge
Build a Huey-based media processing system: thumbnail generation (priority 7, retries 2), metadata extraction (priority 5), format validation (priority 3, retries 1), and cleanup (periodic, every hour). Implement progress tracking via task context and result storage for processed files.
FAQ
Mini Project: Huey Task System
# tasks.py
from huey import RedisHuey, crontab
import time
huey = RedisHuey('demo', host='localhost')
@huey.task(retries=2, retry_delay=5)
def download_file(url):
print(f"Downloading: {url}")
time.sleep(1)
if 'error' in url:
raise Exception("Download failed")
return f"/tmp/{url.split('/')[-1]}"
@huey.task()
def process_file(path):
print(f"Processing: {path}")
time.sleep(2)
return f"{path}.processed"
@huey.task()
def upload_file(path):
print(f"Uploading: {path}")
time.sleep(1)
return f"https://cdn.example.com/{path.split('/')[-1]}"
@huey.periodic_task(crontab(minute='*/5'))
def cleanup():
print("Running cleanup...")
if __name__ == '__main__':
result = (download_file("https://example.com/file.pdf"))
print(f"Download queued: {result.id}")
Expected output:
Download queued: 550e8400-...
What's Next
Now that you understand Huey, explore job scheduling for recurring background tasks, then learn about job priorities for controlling execution order.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro