In this tutorial, we explore how to build a fully functional background task processing system using Huey directly, without relying on Redis. We configure a SQLite-backed Huey instance, start a real consumer in the notebook, and implement advanced task patterns, including retries, priorities, scheduling, pipelines, locking, and monitoring via signals. As we move step by step, we demonstrate how we can simulate production-grade asynchronous job handling while keeping everything self-contained and easy to run in a cloud notebook environment.
!pip -q install -U huey
import os
import time
import json
import random
import threading
from datetime import datetime
from huey import SqliteHuey, crontab
from huey.constants import WORKER_THREAD
DB_PATH = "/content/huey_demo.db"
if os.path.exists(DB_PATH):
os.remove(DB_PATH)
huey = SqliteHuey(
name="colab-huey",
filename=DB_PATH,
results=True,
store_none=False,
utc=True,
)
print("Huey backend:", type(huey).__name__)
print("SQLite DB at:", DB_PATH)We install Huey and configure a SQLite-backed instance. We initialize the database file and ensure a clean environment before starting execution. By doing this, we establish a lightweight yet production-style task queue setup without external dependencies.
EVENT_LOG = []
@huey.signal()
def _log_all_signals(signal, task, exc=None):
EVENT_LOG.append({
"ts": datetime.utcnow().isoformat() + "Z",
"signal": str(signal),
"task": getattr(task, "name", None),
"id": getattr(task, "id", None),
"args": getattr(task, "args", None),
"kwargs": getattr(task, "kwargs", None),
"exc": repr(exc) if exc else None,
})
def print_latest_events(n=10):
print("\n--- Latest Huey events ---")
for row in EVENT_LOG[-n:]:
print(json.dumps(row, indent=2))We implement a signal handler to capture and store task lifecycle events in a structured log. We track execution details, including task IDs, arguments, and exceptions, to improve observability. Through this mechanism, we build real-time monitoring into our asynchronous system.
@huey.task(priority=50)
def quick_add(a, b):
return a + b
@huey.task(priority=10)
def slow_io(seconds=1.0):
time.sleep(seconds)
return f"slept={seconds}"
@huey.task(retries=3, retry_delay=1, priority=100)
def flaky_network_call(p_fail=0.6):
if random.random() < p_fail:
raise RuntimeError("Transient failure (simulated)")
return "OK"
@huey.task(context=True, priority=60)
def cpu_pi_estimate(samples=200_000, task=None):
inside = 0
rnd = random.random
for _ in range(samples):
x, y = rnd(), rnd()
if x*x + y*y <= 1.0:
inside += 1
est = 4.0 * inside / samples
return {"task_id": task.id if task else None, "pi_estimate": est, "samples": samples}We define multiple tasks with priorities, retry configurations, and contextual awareness. We simulate different workloads, including simple arithmetic, I/O delay, transient failures, and CPU-bound computation. By doing this, we demonstrate how Huey handles reliability, execution order, and task metadata.
@huey.lock_task("demo:daily-sync")
@huey.task()
def locked_sync_job(tag="sync"):
time.sleep(1.0)
return f"locked-job-done:{tag}:{datetime.utcnow().isoformat()}Z"
@huey.task()
def fetch_number(seed=7):
random.seed(seed)
return random.randint(1, 100)
@huey.task()
def transform_number(x, scale=3):
return x * scale
@huey.task()
def store_result(x):
return {"stored_value": x, "stored_at": datetime.utcnow().isoformat() + "Z"}We introduce locking to prevent concurrent execution of critical jobs. We also define tasks that will later be chained together using pipelines to form structured workflows. Through this design, we model realistic background processing patterns that require sequencing and concurrency control.
TICK = {"count": 0}
@huey.task()
def heartbeat():
TICK["count"] += 1
print(f"[heartbeat] tick={TICK['count']} utc={datetime.utcnow().isoformat()}Z")
@huey.periodic_task(crontab(minute="*"))
def heartbeat_minutely():
heartbeat()
_TIMER_STATE = {"running": False, "timer": None}
def start_seconds_heartbeat(interval_sec=15):
_TIMER_STATE["running"] = True
def _tick():
if not _TIMER_STATE["running"]:
return
huey.enqueue(heartbeat.s())
t = threading.Timer(interval_sec, _tick)
_TIMER_STATE["timer"] = t
t.start()
_tick()
def stop_seconds_heartbeat():
_TIMER_STATE["running"] = False
t = _TIMER_STATE.get("timer")
if t is not None:
try:
t.cancel()
except Exception:
pass
_TIMER_STATE["timer"] = NoneWe define heartbeat behavior and configure minute-level periodic execution using Huey’s crontab scheduling. We also implement a timer-based mechanism to simulate sub-minute execution intervals for demonstration purposes. With this setup, we create visible recurring background activity within the notebook.
consumer = huey.create_consumer(
workers=4,
worker_type=WORKER_THREAD,
periodic=True,
initial_delay=0.1,
backoff=1.15,
max_delay=2.0,
scheduler_interval=1,
check_worker_health=True,
health_check_interval=10,
flush_locks=False,
)
consumer_thread = threading.Thread(target=consumer.run, daemon=True)
consumer_thread.start()
print("Consumer started (threaded).")
print("\nEnqueue basics...")
r1 = quick_add(10, 32)
r2 = slow_io(0.75)
print("quick_add result:", r1(blocking=True, timeout=5))
print("slow_io result:", r2(blocking=True, timeout=5))
print("\nRetries + priority demo (flaky task)...")
rf = flaky_network_call(p_fail=0.7)
try:
print("flaky_network_call result:", rf(blocking=True, timeout=10))
except Exception as e:
print("flaky_network_call failed even after retries:", repr(e))
print("\nContext task (task id inside payload)...")
rp = cpu_pi_estimate(samples=150_000)
print("pi payload:", rp(blocking=True, timeout=20))
print("\nLocks demo: enqueue multiple locked jobs quickly (should serialize)...")
locked_results = [locked_sync_job(tag=f"run{i}") for i in range(3)]
print([res(blocking=True, timeout=10) for res in locked_results])
print("\nScheduling demo: run slow_io in ~3 seconds...")
rs = slow_io.schedule(args=(0.25,), delay=3)
print("scheduled handle:", rs)
print("scheduled slow_io result:", rs(blocking=True, timeout=10))
print("\nRevoke demo: schedule a task in 5s then revoke before it runs...")
rv = slow_io.schedule(args=(0.1,), delay=5)
rv.revoke()
time.sleep(6)
try:
out = rv(blocking=False)
print("revoked task output:", out)
except Exception as e:
print("revoked task did not produce result (expected):", type(e).__name__, str(e)[:120])
print("\nPipeline demo...")
pipeline = (
fetch_number.s(123)
.then(transform_number, 5)
.then(store_result)
)
pipe_res = huey.enqueue(pipeline)
print("pipeline final result:", pipe_res(blocking=True, timeout=10))
print("\nStarting 15-second heartbeat demo for ~40 seconds...")
start_seconds_heartbeat(interval_sec=15)
time.sleep(40)
stop_seconds_heartbeat()
print("Stopped 15-second heartbeat demo.")
print_latest_events(12)
print("\nStopping consumer gracefully...")
consumer.stop(graceful=True)
consumer_thread.join(timeout=5)
print("Consumer stopped.")We start a threaded consumer inside the notebook to process tasks asynchronously. We enqueue tasks, test retries, demonstrate scheduling and revocation, execute pipelines, and observe logged signals. Finally, we gracefully shut down the consumer to ensure clean resource management and controlled system termination.
In conclusion, we designed and executed an advanced asynchronous task system using Huey with a SQLite backend and an in-notebook consumer. We implemented retries, task prioritization, future scheduling, revocation, locking mechanisms, task chaining through pipelines, and periodic behavior simulation, all within a Colab-friendly setup. Through this approach, we gained a clear understanding of how to use Huey to manage background workloads efficiently and extend this architecture to real-world production deployments.
Check out the Full Coding Notebook/Implementation here. Also, feel free to follow us on Twitter and don’t forget to join our 130k+ ML SubReddit and Subscribe to our Newsletter. Wait! are you on telegram? now you can join us on telegram as well.
Need to partner with us for promoting your GitHub Repo OR Hugging Face Page OR Product Release OR Webinar etc.? Connect with us
Michal Sutter is a data science professional with a Master of Science in Data Science from the University of Padova. With a solid foundation in statistical analysis, machine learning, and data engineering, Michal excels at transforming complex datasets into actionable insights.
此内容由惯性聚合(RSS阅读器)自动聚合整理，仅供阅读参考。 原文来自 — 版权归原作者所有。