Build a lightweight background task queue in Python using asyncio and SQLite WAL mode. Zero external dependencies, excellent performance.
Celery is powerful, but it brings heavy dependencies like Redis or RabbitMQ, extra processes, and significant operational complexity. For many applications, you can build a fast, reliable background task queue using only Python’s native asyncio and SQLite in WAL mode.
This solution is extremely lightweight, requires zero external services, and delivers excellent performance for most use cases.
I replaced Celery in a medium-scale web service with this approach. The system now runs with dramatically lower resource usage and simpler deployment while maintaining reliable task execution.
The Production-Ready Implementation
Here is the complete, robust task queue:
import asyncio
import sqlite3
import json
import time
import uuid
from typing import Callable, Dict, Any, Optional
class LiteQueue:
def __init__(self, db_path: str = "tasks.db", max_retries: int = 3):
self.db_path = db_path
self.max_retries = max_retries
self.conn = sqlite3.connect(db_path, check_same_thread=False)
self.conn.row_factory = sqlite3.Row
self._init_db()
self.running = False
def _init_db(self):
with self.conn:
self.conn.execute("PRAGMA journal_mode=WAL;")
self.conn.execute("PRAGMA synchronous=NORMAL;")
self.conn.execute("PRAGMA cache_size=-20000;")
self.conn.execute("""
CREATE TABLE IF NOT EXISTS tasks (
id TEXT PRIMARY KEY,
task_name TEXT NOT NULL,
payload TEXT NOT NULL,
status TEXT DEFAULT 'pending',
created_at REAL DEFAULT (unixepoch()),
scheduled_at REAL DEFAULT (unixepoch()),
attempts INTEGER DEFAULT 0
)
""")
async def enqueue(self, task_name: str, payload: Dict[str, Any], delay_seconds: int = 0) -> str:
task_id = str(uuid.uuid4())
scheduled_at = time.time() + delay_seconds
payload_str = json.dumps(payload)
def _insert():
with self.conn:
self.conn.execute(
"INSERT INTO tasks (id, task_name, payload, scheduled_at) VALUES (?, ?, ?, ?)",
(task_id, task_name, payload_str, scheduled_at)
)
return task_id
return await asyncio.to_thread(_insert)
async def worker(self, task_handlers: Dict[str, Callable], poll_interval: float = 0.2):
self.running = True
print("🚀 LiteQueue production-ready worker started...")
while self.running:
try:
task = await asyncio.to_thread(self._claim_next_task)
if not task:
await asyncio.sleep(poll_interval)
continue
task_id, task_name, payload_str, attempts = task
payload = json.loads(payload_str)
handler = task_handlers.get(task_name)
if not handler:
print(f"❌ No handler registered for task: {task_name}")
await asyncio.to_thread(self._update_status, task_id, 'failed')
continue
try:
await handler(payload)
await asyncio.to_thread(self._update_status, task_id, 'completed')
except Exception as exc:
print(f"⚠️ Task {task_id} execution failed: {exc}")
await asyncio.to_thread(self._handle_failure, task_id, attempts)
except Exception as e:
print(f"🚨 Worker pipeline anomaly: {e}")
await asyncio.sleep(1)
def _claim_next_task(self) -> Optional[tuple]:
"""Atomically claim the next task to prevent race conditions."""
with self.conn:
cursor = self.conn.execute("""
SELECT id, task_name, payload, attempts
FROM tasks
WHERE status = 'pending' AND scheduled_at <= ?
ORDER BY created_at ASC LIMIT 1
""", (time.time(),))
row = cursor.fetchone()
if not row:
return None
task_id, task_name, payload_str, attempts = row[0], row[1], row[2], row[3]
# Claim-First: Lock the task before yielding control
self.conn.execute(
"UPDATE tasks SET status = 'processing' WHERE id = ? AND status = 'pending'",
(task_id,)
)
return (task_id, task_name, payload_str, attempts)
def _update_status(self, task_id: str, status: str):
with self.conn:
self.conn.execute("UPDATE tasks SET status = ? WHERE id = ?", (status, task_id))
def _handle_failure(self, task_id: str, current_attempts: int):
next_attempts = current_attempts + 1
with self.conn:
if next_attempts >= self.max_retries:
self.conn.execute("UPDATE tasks SET status = 'failed', attempts = ? WHERE id = ?", (next_attempts, task_id))
else:
# Exponential Backoff: Δt = 2^attempts × 10 seconds
backoff_delay = (2 ** current_attempts) * 10
new_scheduled_at = time.time() + backoff_delay
self.conn.execute("""
UPDATE tasks
SET status = 'pending', attempts = ?, scheduled_at = ?
WHERE id = ?
""", (next_attempts, new_scheduled_at, task_id))
def close(self):
self.running = False
self.conn.close()Intelligent Error Resilience with Exponential Backoff
Hard-failing a network-dependent background job on the first exception is poor design. The production version tracks execution attempts and reschedules failures dynamically using an exponential backoff equation:
The Limits of In-Process Queues
Shifting away from Celery is a massive win for stack minimization, but keep these architectural boundaries in mind:
- Horizontal Scaling Hurdles: SQLite is a local file-system database. If your web app grows and you deploy it behind a load balancer across multiple physical servers, each node will only see its own local tasks.db instance.
- Hard Crashes / SIGKILL Vulnerability: If a task state is set to processing and your server encounters a sudden Out-of-Memory event or power loss, that specific task remains stuck in a perpetual processing limbo. You must implement a tiny startup routine to sweep and recover stalled tasks.
Final Thoughts
By shifting security left isn’t a management philosophy — it’s an operational implementation. By forcing your local pre-commit hook to execute zero-disk streaming diff validation against Shannon’s entropy bounds, you create an un-bypassable cryptographic gate. Secrets are intercepted at the memory buffer stage, long before they can ever leak into the history logs of a remote repository.