Local Development for Background Jobs
In this tutorial, you will learn about Local Development for Background Jobs. We cover key concepts, practical examples, and best practices to help you master this topic.
Set up a local development environment for background job processing with Docker Compose, fake workers, inline execution, and debugging tools for faster iteration.
What You Learn
You will learn how to configure local Redis, run workers in development mode, use inline job execution for debugging, and set up Docker Compose for consistent local environments.
Why It Matters
Developing background jobs without proper local setup is slow: no Redis, no worker, no way to debug. A good local dev environment means fast feedback and fewer production surprises.
Real-World Use
DodaTech developers use Docker Compose with Redis, a fake worker that logs jobs instead of processing, and an inline mode that runs jobs synchronously for step-through debugging.
Docker Compose for Local Dev
version: '3.8'
services:
redis:
image: redis:7-alpine
ports:
- "6379:6379"
volumes:
- redis_data:/data
worker:
build: .
command: python worker.py
environment:
- REDIS_URL=redis://redis:6379
- QUEUE_NAME=default
- WORKER_MODE=development
volumes:
- .:/app
depends_on:
- redis
scheduler:
build: .
command: python scheduler.py
environment:
- REDIS_URL=redis://redis:6379
volumes:
- .:/app
depends_on:
- redis
volumes:
redis_data:
Expected output:
Inline Mode for Debugging
import time
import json
class DevJobProcessor:
def __init__(self, inline_mode=False):
self.inline_mode = inline_mode
def enqueue(self, queue, job_data):
if self.inline_mode:
print(f"[DEV] Inline execution of: {job_data.get('task', 'unknown')}")
return self._execute_job(job_data)
else:
print(f"[DEV] Would enqueue to {queue}: {job_data}")
return None
def _execute_job(self, job_data):
task = job_data.get('task')
print(f" Processing: {task}")
result = {'status': 'completed', 'task': task}
if task == 'send_email':
result['to'] = job_data.get('to')
print(f" Email sent to {result['to']}")
elif task == 'scan_file':
result['threats'] = 0
print(f" File scanned: no threats")
else:
print(f" Unknown task: {task}")
return result
# In development, set inline_mode=True
processor = DevJobProcessor(inline_mode=True)
result = processor.enqueue('email', {'task': 'send_email', 'to': 'dev@test.com'})
print(f"Result: {result}")
result = processor.enqueue('scans', {'task': 'scan_file', 'file': 'dev.pdf'})
print(f"Result: {result}")
Expected output:
[DEV] Inline execution of: send_email
Processing: send_email
Email sent to dev@test.com
Result: {'status': 'completed', 'task': 'send_email', 'to': 'dev@test.com'}
[DEV] Inline execution of: scan_file
Processing: scan_file
File scanned: no threats
Result: {'status': 'completed', 'task': 'scan_file', 'threats': 0}
Fake Worker for Development
import json
import time
import threading
class FakeWorker:
def __init__(self, queue='dev_queue'):
self.queue = queue
self.running = True
self.processed_jobs = []
def start(self):
print(f"[FAKE WORKER] Listening on {self.queue}")
while self.running:
time.sleep(2)
self._simulate_work()
def _simulate_work(self):
fake_jobs = [
{'task': 'send_email', 'to': 'user@test.com', 'simulated': True},
{'task': 'generate_report', 'type': 'weekly', 'simulated': True},
{'task': 'cleanup_temp', 'simulated': True},
]
for job in fake_jobs:
self.processed_jobs.append(job)
print(f"[FAKE WORKER] Processed: {job['task']} (simulated)")
def stop(self):
self.running = False
def get_processed(self):
return self.processed_jobs
def summary(self):
return {
'total_processed': len(self.processed_jobs),
'queue': self.queue,
'mode': 'fake_worker',
}
worker = FakeWorker()
t = threading.Thread(target=worker.start, daemon=True)
t.start()
time.sleep(1)
print(f"Summary: {worker.summary()}")
worker.stop()
Expected output:
[FAKE WORKER] Listening on dev_queue
[FAKE WORKER] Processed: send_email (simulated)
[FAKE WORKER] Processed: generate_report (simulated)
[FAKE WORKER] Processed: cleanup_temp (simulated)
Summary: {'total_processed': 3, 'queue': 'dev_queue', 'mode': 'fake_worker'}
Development Mode Config
import os
import json
class DevConfig:
def __init__(self):
self.env = os.getenv('APP_ENV', 'development')
self.inline_jobs = self.env == 'development'
self.redis_url = os.getenv('REDIS_URL', 'redis://localhost:6379')
self.worker_mode = os.getenv('WORKER_MODE', 'thread')
self.log_level = os.getenv('LOG_LEVEL', 'DEBUG' if self.env == 'development' else 'INFO')
self.job_timeout = int(os.getenv('JOB_TIMEOUT', '30'))
self.fake_worker = os.getenv('FAKE_WORKER', 'false').lower() == 'true'
def should_execute_inline(self):
return self.inline_jobs
def get_worker_config(self):
if self.fake_worker:
return {
'type': 'fake',
'queue': 'dev_queue',
'poll_interval': 2,
}
return {
'type': self.worker_mode,
'queue': 'default',
'concurrency': 2 if self.env == 'development' else 8,
'timeout': self.job_timeout,
}
def to_dict(self):
return {
'env': self.env,
'inline_jobs': self.inline_jobs,
'redis_url': self.redis_url,
'worker_mode': self.worker_mode,
'fake_worker': self.fake_worker,
'log_level': self.log_level,
}
config = DevConfig()
print(json.dumps(config.to_dict(), indent=2))
print(f"Worker config: {config.get_worker_config()}")
Expected output:
{
"env": "development",
"inline_jobs": true,
"redis_url": "redis://localhost:6379",
"worker_mode": "thread",
"fake_worker": false,
"log_level": "DEBUG"
}
Worker config: {'type': 'thread', 'queue': 'default', ...}
Common Mistakes
1. Using Production Redis for Development
Development debugging and mistakes affect production data. Always use a separate Redis instance or database for development.
2. No Worker in Dev Environment
Developers who never run workers locally introduce bugs that only appear in production. Always run workers locally.
3. Ignoring Async Behavior
Testing jobs synchronously misses async issues. Even in dev, run jobs through the actual queue occasionally.
4. Hardcoded Connection Strings
Hardcoded Redis URLs break when switching environments. Use environment variables for all connection configuration.
5. No Docker Compose
Different team members have different setups. Docker Compose ensures everyone runs the same Redis version and configuration.
Practice Questions
1. Why use Docker Compose for local job development?
It provides a consistent environment with Redis, worker, and scheduler. No manual installation, same versions for everyone.
2. What is inline job execution?
Jobs run synchronously in the same Process instead of being sent to a queue. Useful for debugging without a running worker.
3. What does a fake worker do?
It simulates job processing without actual execution. Logs what would happen, useful for testing queue flow without real side effects.
4. Why separate development Redis from production?
Prevents development mistakes from corrupting production data. Development Redis can be flushed and restarted freely.
Challenge
Set up a complete local development environment: Docker Compose with Redis, worker, and scheduler, inline mode for debugging, fake worker for integration tests, and environment-based configuration switching.
FAQ
Mini Project: Dev Environment
import os
import json
class LocalDevEnvironment:
def __init__(self):
self.compose = {
'version': '3.8',
'services': {
'redis': {
'image': 'redis:7-alpine',
'ports': ['6379:6379'],
},
'app': {
'build': '.',
'volumes': ['.:/app'],
'environment': [
'REDIS_URL=redis://redis:6379',
'QUEUE_NAME=default',
'WORKER_MODE=inline',
'LOG_LEVEL=DEBUG',
],
'depends_on': ['redis'],
}
}
}
def generate_compose(self):
return json.dumps(self.compose, indent=2)
def start_message(self):
return (
"Local development environment ready:\n"
" - Redis: localhost:6379\n"
" - App: watches /app for changes\n"
" - Mode: inline (jobs run synchronously)\n"
" - Logs: DEBUG level\n"
"\nRun: docker-compose up"
)
env = LocalDevEnvironment()
print(env.start_message())
print("\ndocker-compose.yml:")
print(env.generate_compose()[:200] + "...")
Expected output:
Local development environment ready:
- Redis: localhost:6379
- App: watches /app for changes
- Mode: inline (jobs run synchronously)
- Logs: DEBUG level
Run: docker-compose up
What's Next
Now that you understand local development, explore Docker container jobs for containerized job execution, then learn about Kubernetes jobs for orchestrated processing.
Built by the developers of DodaTech
Doda Browser, DodaZIP & Durga Antivirus Pro