Job Chaining and Sequential Execution
In this tutorial, you will learn about Job Chaining and Sequential Execution. We cover key concepts, practical examples, and best practices to help you master this topic.
Chain Background Jobs sequentially where one job depends on another, with automatic triggering, error propagation, conditional branching, and result passing.
What You Learn
You will learn how to implement job chaining with result passing, conditional branching based on previous job output, error propagation across chains, and timeout handling for long chains.
Why It Matters
Real workflows have multiple steps: validate data, Process it, store results, send notification. Chaining automates this sequence, passing results between steps while handling failures at any point.
Real-World Use
DodaTech's file scanning pipeline chains: download file, scan for malware, generate report, store results, notify user. Each step receives the previous step's output. If scanning fails, the chain stops and notifies the user.
Job Chain Architecture
flowchart TD
A[Job A: Validate] -->|pass data| B[Job B: Process]
B -->|success| C{Decision}
C -->|Condition met| D[Job D: Notify]
C -->|Condition failed| E[Job E: Cleanup]
B -->|error| F[Job F: Error Handler]
D --> G[Complete]
E --> G
F --> H[Alert Admin]
Simple Job Chain
import redis
import json
import time
import uuid
r = redis.Redis()
class JobChain:
def __init__(self, queue='chain_queue'):
self.queue = queue
def start_chain(self, steps, initial_data=None):
chain_id = str(uuid.uuid4())
chain_data = {
'chain_id': chain_id,
'steps': steps,
'current_step': 0,
'results': [],
'data': initial_data or {},
'status': 'running',
}
r.set(f'chain:{chain_id}', json.dumps(chain_data))
self._enqueue_next(chain_id)
return chain_id
def _enqueue_next(self, chain_id):
chain_data = json.loads(r.get(f'chain:{chain_id}'))
if chain_data['current_step'] >= len(chain_data['steps']):
chain_data['status'] = 'completed'
r.set(f'chain:{chain_id}', json.dumps(chain_data))
print(f"Chain {chain_id} completed")
return
step_name = chain_data['steps'][chain_data['current_step']]
job_data = {
'_chain_id': chain_id,
'_step': chain_data['current_step'],
'_step_name': step_name,
'_chain_data': chain_data['data'],
}
r.lpush(f'chain:{chain_id}:jobs', json.dumps(job_data))
r.lpush(self.queue, json.dumps({
'type': 'chain_step',
'chain_id': chain_id,
'step': chain_data['current_step'],
'step_name': step_name,
}))
print(f"Enqueued step {chain_data['current_step']}: {step_name}")
def complete_step(self, chain_id, step_result):
chain_data = json.loads(r.get(f'chain:{chain_id}'))
chain_data['results'].append(step_result)
chain_data['data'].update(step_result.get('data', {}))
chain_data['current_step'] += 1
r.set(f'chain:{chain_id}', json.dumps(chain_data))
self._enqueue_next(chain_id)
def fail_step(self, chain_id, error):
chain_data = json.loads(r.get(f'chain:{chain_id}'))
chain_data['status'] = 'failed'
chain_data['error'] = error
r.set(f'chain:{chain_id}', json.dumps(chain_data))
print(f"Chain {chain_id} failed at step {chain_data['current_step']}: {error}")
def get_chain_status(self, chain_id):
data = r.get(f'chain:{chain_id}')
return json.loads(data) if data else None
chain = JobChain()
cid = chain.start_chain(['validate', 'process', 'notify'], {'file': 'report.pdf'})
chain.complete_step(cid, {'status': 'ok', 'data': {'validated': True}})
chain.complete_step(cid, {'status': 'ok', 'data': {'processed': 'done'}})
status = chain.get_chain_status(cid)
print(f"Chain status: {status['status']}")
Expected output:
Enqueued step 0: validate
Chain d4e5f6... completed
Chain status: completed
Conditional Chain Branching
import redis
import json
import time
r = redis.Redis()
class ConditionalChain:
def __init__(self):
self.chains = {}
def define_chain(self, name, steps, conditions=None):
self.chains[name] = {
'steps': steps,
'conditions': conditions or {},
}
def execute(self, chain_name, initial_data=None):
chain_def = self.chains.get(chain_name)
if not chain_def:
raise ValueError(f"Unknown chain: {chain_name}")
chain_id = f"chain-{time.time_ns()}"
state = {
'chain_id': chain_id,
'chain_name': chain_name,
'step_index': 0,
'data': initial_data or {},
'results': [],
'status': 'running',
}
r.set(f'chain_state:{chain_id}', json.dumps(state))
self._run_step(chain_id)
return chain_id
def _run_step(self, chain_id):
state = json.loads(r.get(f'chain_state:{chain_id}'))
chain_def = self.chains[state['chain_name']]
if state['step_index'] >= len(chain_def['steps']):
state['status'] = 'completed'
r.set(f'chain_state:{chain_id}', json.dumps(state))
print(f"Chain {chain_id} completed successfully")
return
step = chain_def['steps'][state['step_index']]
conditions = chain_def['conditions'].get(state['step_index'], {})
for cond_key, cond_value in conditions.items():
if state['data'].get(cond_key) == cond_value:
target_step = cond_value.get('skip_to')
if target_step is not None:
state['step_index'] = target_step
r.set(f'chain_state:{chain_id}', json.dumps(state))
self._run_step(chain_id)
return
print(f"Running step {state['step_index']}: {step}")
state['step_index'] += 1
r.set(f'chain_state:{chain_id}', json.dumps(state))
def feed_result(self, chain_id, result):
state = json.loads(r.get(f'chain_state:{chain_id}'))
state['data'].update(result)
state['results'].append(result)
r.set(f'chain_state:{chain_id}', json.dumps(state))
self._run_step(chain_id)
cc = ConditionalChain()
cc.define_chain('file_pipeline', ['validate', 'scan', 'report', 'notify'], {
1: {'needs_full_scan': True},
})
cid = cc.execute('file_pipeline', {'filename': 'doc.pdf'})
cc.feed_result(cid, {'validated': True})
cc.feed_result(cid, {'threats_found': False})
cc.feed_result(cid, {'report_id': 'rpt-001'})
Expected output:
Running step 0: validate
Running step 1: scan
Running step 2: report
Running step 3: notify
Chain ... completed successfully
Result Passing Between Steps
import json
import time
class ResultPassingChain:
def __init__(self):
self.handlers = {}
self.chain_data = {}
def step(self, name):
def decorator(func):
self.handlers[name] = func
return func
return decorator
def run(self, steps, initial=None):
data = initial or {}
results = []
for step_name in steps:
handler = self.handlers.get(step_name)
if not handler:
raise ValueError(f"No handler for step: {step_name}")
print(f"Step: {step_name}")
result = handler(data)
data.update(result)
results.append(result)
return results
chain = ResultPassingChain()
@chain.step('download')
def download(data):
print(f" Downloading {data['url']}")
return {'local_path': '/tmp/file.pdf', 'size_bytes': 1024000}
@chain.step('scan')
def scan(data):
print(f" Scanning {data['local_path']}")
return {'threats': 0, 'scan_time_ms': 150}
@chain.step('report')
def report(data):
print(f" Generating report: {data['local_path']}")
return {'report_url': '/reports/file.pdf'}
results = chain.run(['download', 'scan', 'report'], {'url': 'https://example.com/file.pdf'})
for r in results:
print(f" Result: {r}")
Expected output:
Step: download
Downloading https://example.com/file.pdf
Step: scan
Scanning /tmp/file.pdf
Step: report
Generating report: /tmp/file.pdf
Result: {'local_path': '/tmp/file.pdf', 'size_bytes': 1024000}
Result: {'threats': 0, 'scan_time_ms': 150}
Result: {'report_url': '/reports/file.pdf'}
Error Propagation in Chains
import time
class ChainWithErrorHandling:
def __init__(self):
self.handlers = {}
self.error_handlers = {}
def step(self, name):
def decorator(func):
self.handlers[name] = func
return func
return decorator
def on_error(self, step_name):
def decorator(func):
self.error_handlers[step_name] = func
return func
return decorator
def run(self, steps, initial=None):
data = initial or {}
for step_name in steps:
handler = self.handlers.get(step_name)
if not handler:
continue
try:
print(f"Running: {step_name}")
result = handler(data)
data.update(result)
except Exception as e:
error_handler = self.error_handlers.get(step_name)
if error_handler:
print(f"Error in {step_name}: {e}")
error_handler(data, e)
else:
print(f"Unhandled error in {step_name}: {e}")
return data
return data
ceh = ChainWithErrorHandling()
@ceh.step('validate')
def validate(data):
print(" Validation passed")
return {'valid': True}
@ceh.step('process')
def process(data):
raise ValueError("Processing failed: corrupt data")
@ceh.step('notify')
def notify(data):
return {'notified': True}
@ceh.on_error('process')
def handle_process_error(data, error):
print(f" Error handler: {error}")
data['error_handled'] = True
result = ceh.run(['validate', 'process', 'notify'])
print(f"Final data: {result}")
Expected output:
Running: validate
Validation passed
Running: process
Error in process: Processing failed: corrupt data
Error handler: Processing failed: corrupt data
Final data: {'valid': True, 'error_handled': True}
Common Mistakes
1. No Error Propagation
When a step fails, remaining steps continue with bad data or the chain silently stops. Define error handlers per step and propagate errors.
2. Tight Coupling Between Steps
Steps that share mutable state or call each other directly defeat the purpose of chaining. Each step should be independent and receive only its input data.
3. Missing Timeout for Chains
A stuck step blocks the entire chain forever. Implement per-step timeouts and chain-level timeout with automatic failure.
4. No Idempotency for Chained Jobs
If a step completes but acknowledgment fails, the chain retries from that step. Design idempotent steps that handle re-execution safely.
5. Circular Dependencies
Chain definitions that loop forever (A depends on B, B depends on A) cause infinite processing. Validate chains for cycles before execution.
Practice Questions
1. How does job chaining differ from simple sequential execution?
Chaining passes results between steps, supports conditional branching, and includes error propagation. Sequential execution runs steps independently without data flow.
2. What is conditional branching in a chain?
Based on the result of one step, the chain decides which step to run next. Example: if scan finds threats, run quarantine step instead of notify.
3. How do you handle failures in a chain?
Define error handlers per step. On failure, the error handler runs (cleanup, notification) and the chain stops or takes an alternative path.
4. Why is idempotency important in chained jobs?
If a step completes but the acknowledgment is lost, the chain retries from that step. Idempotent steps handle re-execution without side effects.
Challenge
Build a job chain for document processing: download document, validate format (PDF/DOCX), extract text, translate if needed, generate summary, store results, notify user. Include conditional branching for translation and error handling at each step.
FAQ
Mini Project: Document Processing Chain
import time
import json
class DocProcessingChain:
def __init__(self):
self.steps = {}
self.results = {}
def register(self, name):
def decorator(func):
self.steps[name] = func
return func
return decorator
def process(self, doc_id, pipeline, initial_data=None):
data = initial_data or {}
for step_name in pipeline:
step_func = self.steps.get(step_name)
if not step_func:
print(f"Unknown step: {step_name}")
break
print(f"[{doc_id}] {step_name}...")
try:
result = step_func(data)
data.update(result)
self.results[f"{doc_id}:{step_name}"] = result
except Exception as e:
print(f"[{doc_id}] {step_name} FAILED: {e}")
break
return data
dpc = DocProcessingChain()
@dpc.register('validate')
def validate(data):
if not data.get('filename', '').endswith('.pdf'):
raise ValueError("Only PDF files supported")
return {'valid': True}
@dpc.register('extract')
def extract(data):
return {'pages': 12, 'characters': 45000}
@dpc.register('scan')
def scan(data):
return {'threats': 0, 'safe': True}
@dpc.register('store')
def store(data):
return {'stored_at': f"/docs/{data['filename']}"}
result = dpc.process('doc-1', ['validate', 'extract', 'scan', 'store'],
{'filename': 'report.pdf'})
print(f"Final: {result}")
Expected output:
[doc-1] validate...
[doc-1] extract...
[doc-1] scan...
[doc-1] store...
Final: {'filename': 'report.pdf', 'valid': True, 'pages': 12, ...}
What's Next
Now that you understand job chaining, explore job DAG workflows for complex Orchestration, then learn about job progress websocket for real-time chain status.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro