A Coding Guide to Build a Production-Grade Background Task Processing System Using Huey with SQLite, Scheduling, Retries, Pipelines, and Concurrency Control

!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)

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))

@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}

@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"}

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"] = None

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.")