Progress Tracking for Background Jobs
In this tutorial, you will learn about Progress Tracking for Background Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Track progress of long-running background jobs with percentage updates, status reporting, and real-time monitoring for better user experience.
What You Learn
You will learn how to implement progress tracking for jobs, update progress from workers, poll progress from clients, and display progress in web UIs.
Why It Matters
Long-running jobs without progress feedback feel broken. Users do not know if the job is still running, stuck, or failed. Progress tracking provides visibility and improves user confidence in the system.
Real-World Use
Doda Browser's file analysis shows real-time progress: "Scanning file (25%)", "Analyzing threats (50%)", "Generating report (75%)", "Complete (100%)". Users can see the Process moving forward.
Progress Store
import redis
import json
import time
r = redis.Redis()
class ProgressStore:
def __init__(self):
self.prefix = 'progress:'
def set_progress(self, job_id, percentage, status='running', message=''):
data = {
'percentage': percentage,
'status': status,
'message': message,
'updated_at': time.time(),
}
r.setex(f"{self.prefix}{job_id}", 3600, json.dumps(data))
def get_progress(self, job_id):
data = r.get(f"{self.prefix}{job_id}")
return json.loads(data) if data else None
def complete(self, job_id, result=None):
progress = self.get_progress(job_id) or {}
progress.update({
'percentage': 100,
'status': 'completed',
'result': result,
'updated_at': time.time(),
})
r.setex(f"{self.prefix}{job_id}", 3600, json.dumps(progress))
def fail(self, job_id, error):
progress = self.get_progress(job_id) or {}
progress.update({
'percentage': 100,
'status': 'failed',
'error': error,
'updated_at': time.time(),
})
r.setex(f"{self.prefix}{job_id}", 3600, json.dumps(progress))
store = ProgressStore()
store.set_progress('job-1', 25, 'running', 'Processing batch 1/4')
time.sleep(1)
print(store.get_progress('job-1'))
store.complete('job-1', {'output': 'report.pdf'})
print(store.get_progress('job-1'))
Expected output:
{'percentage': 25, 'status': 'running', 'message': 'Processing batch 1/4', 'updated_at': ...}
{'percentage': 100, 'status': 'completed', 'result': {'output': 'report.pdf'}, ...}
Worker with Progress
import redis
import json
import time
r = redis.Redis()
class ProgressWorker:
def __init__(self):
self.progress = ProgressStore()
def process_video(self, job_id, video_path, steps):
total_steps = len(steps)
print(f"Starting video processing: {job_id}")
for i, step in enumerate(steps):
percentage = int(((i + 1) / total_steps) * 100)
self.progress.set_progress(job_id, percentage, 'running',
f"{step['name']} ({percentage}%)")
print(f" {step['name']}...")
time.sleep(step.get('duration', 1))
print(f" {step['name']} complete")
self.progress.complete(job_id, {'url': f"/videos/{video_path}.mp4"})
print(f"Video processing complete: {job_id}")
def poll_progress(self, job_id):
while True:
progress = self.progress.get_progress(job_id)
if not progress:
print("Job not found")
break
print(f"[{progress['percentage']}%] {progress['message']}")
if progress['status'] in ('completed', 'failed'):
print(f"Status: {progress['status']}")
if 'result' in progress:
print(f"Result: {progress['result']}")
break
time.sleep(0.5)
worker = ProgressWorker()
steps = [
{'name': 'Validating file', 'duration': 0.5},
{'name': 'Transcoding video', 'duration': 2},
{'name': 'Generating thumbnails', 'duration': 1},
{'name': 'Uploading result', 'duration': 0.5},
]
import threading
t = threading.Thread(target=worker.process_video, args=('vid-1', 'intro', steps), daemon=True)
t.start()
worker.poll_progress('vid-1')
Expected output:
Starting video processing: vid-1
[25%] Transcoding video (25%)
Validating file...
Validating file complete
[50%] Transcoding video (50%)
Transcoding video...
...
[100%] Completed
Status: completed
Result: {'url': '/videos/intro.mp4'}
Web UI Polling
import redis
import json
import time
r = redis.Redis()
class ProgressAPI:
def __init__(self):
self.store = ProgressStore()
def get_status(self, job_id):
progress = self.store.get_progress(job_id)
if not progress:
return {'status': 'not_found', 'job_id': job_id}
return {
'job_id': job_id,
'percentage': progress['percentage'],
'status': progress['status'],
'message': progress.get('message', ''),
'updated_at': progress.get('updated_at'),
'result': progress.get('result'),
'error': progress.get('error'),
}
def get_multiple(self, job_ids):
return [self.get_status(jid) for jid in job_ids]
api = ProgressAPI()
# Simulate progress updates
for pct in [0, 25, 50, 75, 100]:
status = 'completed' if pct == 100 else 'running'
r.setex(f"progress:job-a", 3600, json.dumps({
'percentage': pct,
'status': status,
'message': f'Processing ({pct}%)',
}))
print(json.dumps(api.get_status('job-a'), indent=2))
Expected output:
{
"job_id": "job-a",
"percentage": 100,
"status": "completed",
"message": "Processing (100%)",
...
}
Websocket Progress
import redis
import json
import time
import threading
r = redis.Redis()
class WebSocketProgress:
def __init__(self):
self.channel = 'progress_updates'
def publish_progress(self, job_id, percentage, message):
data = json.dumps({
'job_id': job_id,
'percentage': percentage,
'message': message,
'timestamp': time.time(),
})
r.publish(self.channel, data)
def subscribe(self, callback):
pubsub = r.pubsub()
pubsub.subscribe(self.channel)
for message in pubsub.listen():
if message['type'] == 'message':
callback(json.loads(message['data']))
def on_progress(data):
print(f"[{data['percentage']}%] {data['message']}")
ws = WebSocketProgress()
t = threading.Thread(target=ws.subscribe, args=(on_progress,), daemon=True)
t.start()
time.sleep(0.5)
ws.publish_progress('job-1', 25, 'Processing batch 1')
ws.publish_progress('job-1', 50, 'Processing batch 2')
ws.publish_progress('job-1', 100, 'Complete')
time.sleep(1)
Expected output:
[25%] Processing batch 1
[50%] Processing batch 2
[100%] Complete
Common Mistakes
1. Updating Progress Too Frequently
Updating progress on every iteration creates Redis write storms. Update at most every 1-2 seconds or every 5% change.
2. Not Setting Progress TTL
Progress records accumulate in Redis. Set TTL to auto-cleanup old progress data. Match TTL to expected job duration plus a buffer.
3. Blocking the Worker for Progress Updates
Progress updates should be non-blocking. If Redis is slow, the worker should not pause processing to update progress.
4. Not Handling Unknown Job IDs
Polling a non-existent job ID should return a clear status (not_found) rather than erroring.
5. Ignoring Progress on the Client Side
Progress tracking only helps if the UI displays it. Always show progress bars or status indicators to users.
Practice Questions
1. How do you update job progress from a worker?
Store percentage and status in Redis with the job ID as key. Workers update progress at meaningful milestones.
2. How does the client poll for progress?
The client periodically calls an API endpoint that reads the progress from Redis and returns it. Use setInterval or WebSocket for real-time updates.
3. What is a good progress update frequency?
Every 1-2 seconds or every 5-10% progress. Too frequent updates waste resources. Too infrequent updates frustrate users.
4. How long should progress data be stored?
Match the TTL to the maximum expected job duration plus a buffer. For most jobs, 1-2 hours is sufficient.
Challenge
Design a progress tracking system for a document processing pipeline with 5 steps: upload (0-10%), format validation (10-25%), OCR extraction (25-60%), analysis (60-85%), report generation (85-100%). Implement both polling and WebSocket updates, show estimated time remaining, and handle job cancellation mid-progress.
FAQ
Mini Project: Progress Tracker
import redis
import json
import time
import threading
r = redis.Redis()
class JobProgress:
def __init__(self, job_id, total_steps):
self.job_id = job_id
self.total = total_steps
self.current = 0
def advance(self, step_name):
self.current += 1
pct = int((self.current / self.total) * 100)
data = {
'percentage': pct,
'current_step': self.current,
'total_steps': self.total,
'step_name': step_name,
'status': 'running',
'updated_at': time.time(),
}
r.setex(f"job_progress:{self.job_id}", 3600, json.dumps(data))
return data
def complete(self, result=None):
data = {
'percentage': 100,
'current_step': self.total,
'total_steps': self.total,
'step_name': 'Complete',
'status': 'completed',
'result': result,
'updated_at': time.time(),
}
r.setex(f"job_progress:{self.job_id}", 3600, json.dumps(data))
def fail(self, error):
data = {
'percentage': int((self.current / self.total) * 100),
'status': 'failed',
'error': error,
'updated_at': time.time(),
}
r.setex(f"job_progress:{self.job_id}", 3600, json.dumps(data))
@staticmethod
def get(job_id):
data = r.get(f"job_progress:{job_id}")
return json.loads(data) if data else None
def process_files(job_id, files):
steps = ['validate', 'scan', 'analyze', 'report']
prog = JobProgress(job_id, len(files) * len(steps))
for f in files:
for step in steps:
prog.advance(f"{step}:{f}")
time.sleep(0.3)
prog.complete({'processed': len(files)})
print(f"Job {job_id} complete")
def monitor(job_id):
while True:
p = JobProgress.get(job_id)
if p:
print(f" [{p['percentage']}%] {p.get('step_name', '')}")
if p['status'] in ('completed', 'failed'):
break
time.sleep(0.5)
t = threading.Thread(target=process_files, args=('batch-1', ['a.txt', 'b.txt']), daemon=True)
t.start()
monitor('batch-1')
Expected output:
[12%] validate:a.txt
[25%] scan:a.txt
[37%] analyze:a.txt
...
[100%] Complete
Job batch-1 complete
What's Next
Now that you understand progress tracking, explore job cancellation mechanisms, then learn about recurring jobs for periodic execution.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro