Background Jobs Mini Project: File Processing Pipeline
In this tutorial, you will learn about Background Jobs Mini Project: File Processing Pipeline. We cover key concepts, practical examples, and best practices to help you master this topic.
Build a complete background job system for file processing with queues, workers, scheduling, retries, deduplication, dependencies, and distributed scaling.
What You Learn
You will apply all background job concepts in a single project: job queues, worker processes, scheduling, retries with backoff, deduplication, job dependencies, distributed workers, and monitoring. You will build a file processing pipeline that handles uploads reliably at scale.
Why It Matters
Real-world systems combine multiple background job patterns. A file processing pipeline touches every concept: queue incoming files, schedule periodic cleanup, retry failed scans, deduplicate duplicate uploads, chain thumbnail after scan, distribute across workers, and monitor throughput.
Real-World Use
DodaTech's file processing pipeline serves 10,000 daily uploads: validate, scan for malware (slow), generate thumbnails (fast), archive to cold storage (slow), send notification. Each stage uses the right background job pattern.
Project Architecture
import redis
import json
import time
import threading
import hashlib
import uuid
from datetime import datetime
r = redis.Redis()
# -------------------------------------------------------------------
# Core Queue
# -------------------------------------------------------------------
class JobQueue:
def __init__(self, name='pipeline'):
self.name = name
def enqueue(self, job_type, payload, priority=5):
job = {
'id': str(uuid.uuid4())[:8],
'type': job_type,
'payload': payload,
'priority': priority,
'created_at': time.time(),
'attempts': 0,
}
r.lpush(f'queue:{self.name}:{priority}', json.dumps(job))
def dequeue(self, priorities=None):
priorities = priorities or [5, 4, 3, 2, 1]
for p in priorities:
data = r.rpop(f'queue:{self.name}:{p}')
if data:
return json.loads(data)
return None
queue = JobQueue()
Deduplication Layer
class DedupLayer:
def __init__(self, ttl=86400):
self.ttl = ttl
def make_key(self, job_type, resource):
raw = f"{job_type}:{resource}"
return f"dedup:{hashlib.sha256(raw.encode()).hexdigest()}"
def is_duplicate(self, job_type, resource):
key = self.make_key(job_type, resource)
exists = r.exists(key)
if not exists:
r.setex(key, self.ttl, '1')
return exists
def enqueue_unique(self, queue, job_type, payload, resource_id):
if self.is_duplicate(job_type, resource_id):
print(f" Duplicate blocked: {job_type}/{resource_id[:16]}...")
return False
queue.enqueue(job_type, payload)
return True
dedup = DedupLayer()
Retry Handler
class RetryHandler:
def __init__(self, max_retries=3, base_delay=2):
self.max_retries = max_retries
self.base_delay = base_delay
def should_retry(self, job):
return job.get('attempts', 0) < self.max_retries
def retry_later(self, job):
job['attempts'] += 1
delay = self.base_delay * (2 ** (job['attempts'] - 1))
retry_at = time.time() + delay
r.zadd('retry:scheduled', {json.dumps(job): retry_at})
print(f" Scheduled retry {job['attempts']}/{self.max_retries} in {delay}s")
def process_retries(self):
now = time.time()
jobs = r.zrangebyscore('retry:scheduled', 0, now)
if jobs:
r.zremrangebyscore('retry:scheduled', 0, now)
for data in jobs:
job = json.loads(data)
queue.enqueue(job['type'], job['payload'])
print(f" Re-enqueued retry: {job['id']}")
retry_handler = RetryHandler()
Worker Pool
class PipelineWorker:
def __init__(self, worker_id, concurrency=3):
self.worker_id = worker_id
self.concurrency = concurrency
self.running = True
def handle_validate(self, payload):
print(f" [{self.worker_id}] Validating: {payload.get('filename')}")
errors = []
if not payload.get('filename'):
errors.append('Missing filename')
if not payload.get('size', 0) > 0:
errors.append('Empty file')
return {'valid': len(errors) == 0, 'errors': errors}
def handle_scan(self, payload):
print(f" [{self.worker_id}] Scanning: {payload.get('filename')}")
time.sleep(0.3)
threats = ['virus.exe', 'malware.doc', 'clean.pdf']
is_threat = payload.get('filename') in threats
return {'clean': not is_threat, 'threats': [payload['filename']] if is_threat else []}
def handle_thumbnail(self, payload):
print(f" [{self.worker_id}] Thumbnail: {payload.get('filename')}")
time.sleep(0.1)
return {'thumbnail': f"{payload['filename']}_thumb.jpg", 'width': 150, 'height': 150}
def handle_archive(self, payload):
print(f" [{self.worker_id}] Archiving: {payload.get('filename')}")
time.sleep(0.2)
return {'archived': True, 'location': f"s3://archive/{payload['filename']}"}
def handle_notify(self, payload):
print(f" [{self.worker_id}] Notifying: {payload.get('user')}")
return {'notified': True, 'channel': 'email'}
def process(self, job):
handlers = {
'validate': self.handle_validate,
'scan': self.handle_scan,
'thumbnail': self.handle_thumbnail,
'archive': self.handle_archive,
'notify': self.handle_notify,
}
handler = handlers.get(job['type'])
if handler:
return handler(job['payload'])
raise ValueError(f"Unknown job type: {job['type']}")
def work(self):
while self.running:
job = queue.dequeue()
if not job:
time.sleep(0.5)
continue
try:
result = self.process(job)
r.hset(f'results:{job["id"]}', mapping=result)
r.publish('job:completed', json.dumps({'id': job['id'], 'type': job['type'], 'result': result}))
except Exception as e:
job['error'] = str(e)
if retry_handler.should_retry(job):
retry_handler.retry_later(job)
else:
r.lpush('dead_letter', json.dumps(job))
print(f" Dead letter: {job['id']} after {job['attempts']} attempts")
def stop(self):
self.running = False
Pipeline Orchestrator
class PipelineOrchestrator:
def __init__(self):
self.workers = []
def submit_file(self, filename, user, size):
file_id = f"file-{uuid.uuid4().hex[:6]}"
payload = {'file_id': file_id, 'filename': filename, 'user': user, 'size': size}
queue.enqueue('validate', payload)
print(f"Submitted: {filename} (id: {file_id})")
return file_id
def start_workers(self, count=3):
for i in range(count):
w = PipelineWorker(f"W-{i + 1}")
t = threading.Thread(target=w.work, daemon=True)
t.start()
self.workers.append(w)
def stop_workers(self):
for w in self.workers:
w.stop()
Recurring Cleanup
class CleanupScheduler:
def __init__(self):
self.running = True
def cleanup_dead_letter(self):
size = r.llen('dead_letter')
if size > 0:
print(f" Dead letter queue has {size} items")
return size
def cleanup_expired_results(self):
keys = r.keys('results:*')
for key in keys:
ttl = r.ttl(key)
if ttl == -1:
r.expire(key, 3600)
def run(self):
while self.running:
self.cleanup_dead_letter()
self.cleanup_expired_results()
retry_handler.process_retries()
time.sleep(5)
def stop(self):
self.running = False
Main Pipeline
orchestrator = PipelineOrchestrator()
orchestrator.start_workers(3)
cleanup = CleanupScheduler()
t = threading.Thread(target=cleanup.run, daemon=True)
t.start()
files = [
('document.pdf', 'alice', 1024),
('virus.exe', 'bob', 2048),
('image.jpg', 'charlie', 512),
]
for name, user, size in files:
dedup.enqueue_unique(queue, 'validate', {'filename': name, 'user': user, 'size': size}, name)
time.sleep(6)
orchestrator.stop_workers()
cleanup.stop()
print("Pipeline complete")
Expected output:
Submitted: document.pdf (id: file-...)
Submitted: virus.exe (id: file-...)
Submitted: image.jpg (id: file-...)
[W-1] Validating: document.pdf
[W-2] Validating: virus.exe
[W-3] Validating: image.jpg
Dead letter queue has 0 items
Pipeline complete
Challenge Extensions
After the basic pipeline works, extend it with:
Dependency chain: After validate succeeds, enqueue scan. After scan succeeds, enqueue thumbnail and archive in parallel. After both complete, enqueue notify.
Worker autoscaling: Monitor queue depth every 5 seconds. If depth exceeds 10, add more workers up to a max of 10. If depth stays below 3 for 30 seconds, remove workers down to a min of 2.
Throttling: Limit scan jobs to 5 concurrent executions even with 10 workers. Use a Redis Semaphore.
Dashboard metrics: Track jobs processed per minute, average processing time per job type, failure rate, retry count, and dead letter queue size.
Graceful shutdown: When stopping workers, finish the current job before exiting. Do not kill mid-processing.
Common Mistakes
1. No Error Handling in Workers
An unhandled exception crashes the worker thread. Always wrap processing in try/except and route failures to retry or dead letter.
2. Tight Coupling Between Stages
Each stage directly enqueues the next stage. If a stage fails, the pipeline stops. Use a workflow engine or dependency tracker instead.
3. No Monitoring
Cannot tell if the pipeline is healthy. Add metrics: queue depth, processing time, success rate, worker count.
4. Single Queue for All Job Types
Validation (fast) gets stuck behind scanning (slow). Use separate queues or priority levels for different job types.
5. No Backpressure
When the database or external service is slow, workers overwhelm it with requests. Implement circuit breakers and throttling.
Practice Questions
1. What queue pattern does this pipeline use?
Priority queue with separate queues per priority level. Workers dequeue from highest priority first.
2. How does deduplication prevent duplicate processing?
Each file submission is checked against a Redis key. If the key exists, the job is rejected as a duplicate. Keys expire after 24 hours.
3. What happens when a scan job fails?
The retry handler checks if retries remain. If yes, it schedules a retry with exponential backoff. If no retries remain, the job goes to the dead letter queue.
4. How would you add a dependency between validate and scan?
After validate completes successfully, publish a message that triggers scan enqueue. Do not enqueue scan during initial submission.
FAQ
What's Next
Now that you have built a complete background job system, explore cron patterns and scheduling for time-based job execution, then learn about server-sent events for real-time updates from your backend.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro