Scheduling Python Jobs with APScheduler and Redis

· 10 min read ·
Scheduling Python Jobs with APScheduler and Redis

What You’ll Need

  • n8n Cloud or self-hosted n8n for workflow orchestration
  • Hetzner VPS or Contabo VPS for hosting your scheduler
  • Python 3.8 or higher installed locally
  • Redis server (local or cloud instance)
  • Basic familiarity with Python and command line tools

Table of Contents

Understanding APScheduler and Redis

I’ve spent the last few years building distributed job schedulers for various platforms, and I can tell you that the combination of APScheduler and Redis is one of the most reliable approaches for production workloads. APScheduler (Advanced Python Scheduler) is a robust library that lets you schedule Python functions to run at specific times, intervals, or based on cron expressions. Redis, on the other hand, serves as a distributed data store that allows multiple instances of your application to coordinate and share job state.

The magic happens when you use Redis as APScheduler’s job store. Instead of keeping jobs in memory on a single machine, they’re persisted in Redis, which means your scheduled tasks survive application restarts and can be managed across multiple worker instances. This is critical for any serious production system.

Without Redis, if your scheduler crashes, you lose track of what jobs should have run. With Redis, that state persists, and when your scheduler comes back online, it catches up on missed executions. This is the foundation of reliable task scheduling at scale.

Setting Up Your Environment

First, let’s get the basics in place. I’m going to walk you through setting up APScheduler with Redis step by step, and we’ll build toward a complete production-ready example.

Start by creating a new Python project directory and installing the required packages:

mkdir apscheduler-redis-demo
cd apscheduler-redis-demo
python -m venv venv
source venv/bin/activate
pip install apscheduler redis flask

On Windows, activate the virtual environment with venv\Scripts\activate instead.

Next, you’ll need Redis running. If you’re on a local development machine, you can use Docker:

docker run -d -p 6379:6379 redis:7-alpine

Or if you have Redis installed locally, just start the server:

redis-server

Verify the connection works by opening a Python shell and testing:

import redis
r = redis.Redis(host='localhost', port=6379, db=0)
r.ping()

If you see True returned, you’re connected and ready to go.

Building Your First Scheduled Job

Let me show you how to set up a basic scheduler. Create a file called scheduler.py:

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.executors.pool import ThreadPoolExecutor, ProcessPoolExecutor
import redis
import logging
from datetime import datetime

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

redis_conn = redis.Redis(host='localhost', port=6379, db=0)

jobstores = {
    'default': RedisJobStore(connection=redis_conn)
}

executors = {
    'default': ThreadPoolExecutor(max_workers=20),
    'processpool': ProcessPoolExecutor(max_workers=5)
}

job_defaults = {
    'coalesce': False,
    'max_instances': 1
}

scheduler = BackgroundScheduler(
    jobstores=jobstores,
    executors=executors,
    job_defaults=job_defaults,
    timezone='UTC'
)

def my_scheduled_job(name):
    logger.info(f'Job executed at {datetime.now()}: Hello {name}!')

def setup_jobs():
    scheduler.add_job(
        my_scheduled_job,
        'interval',
        seconds=10,
        args=['World'],
        id='job_every_10_seconds',
        name='Hello World Job',
        replace_existing=True
    )
    
    scheduler.add_job(
        my_scheduled_job,
        'cron',
        hour=12,
        minute=0,
        args=['Noon'],
        id='job_at_noon',
        name='Noon Job',
        replace_existing=True
    )
    
    scheduler.add_job(
        my_scheduled_job,
        'interval',
        minutes=5,
        args=['Every 5 Minutes'],
        id='job_every_5_min',
        name='Every 5 Minutes Job',
        replace_existing=True
    )

if __name__ == '__main__':
    setup_jobs()
    scheduler.start()
    
    logger.info('Scheduler started successfully')
    
    try:
        while True:
            pass
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        logger.info('Scheduler shut down')

This is your foundation. The scheduler connects to Redis using RedisJobStore, which persists all job definitions and execution state. The jobstores dictionary maps job store names to their configurations, and the executors determine how jobs actually run (threading, multiprocessing, etc.).

The job_defaults are important: coalesce: False means if a job misses its scheduled time (due to downtime), it won’t try to catch up by running multiple times at once. max_instances: 1 prevents the same job from running concurrently.

Now let’s run this and verify it works:

python scheduler.py

You should see log output showing the scheduler started. Check Redis to see your jobs stored there:

redis-cli
> KEYS *

You’ll see keys like apscheduler.jobs, apscheduler.last_tick_time, etc. This confirms your jobs are persisted.

💡 Fast-Track Your Project: Don’t want to configure this yourself? I build custom n8n pipelines and bots. Message me with code SYS3-HUGO.

Scaling with Redis as a Job Store

Here’s where things get interesting. Let me show you how to create a distributed scheduler setup where multiple instances can coordinate through Redis. Create a file called distributed_scheduler.py:

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.triggers.cron import CronTrigger
import redis
import logging
import os
import socket
from datetime import datetime
import json

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

REDIS_HOST = os.getenv('REDIS_HOST', 'localhost')
REDIS_PORT = int(os.getenv('REDIS_PORT', 6379))
REDIS_DB = int(os.getenv('REDIS_DB', 0))

redis_conn = redis.Redis(
    host=REDIS_HOST,
    port=REDIS_PORT,
    db=REDIS_DB,
    decode_responses=True
)

hostname = socket.gethostname()

jobstores = {
    'default': RedisJobStore(
        connection=redis_conn,
        jobs_key='apscheduler.jobs',
        run_times_key='apscheduler.run_times'
    )
}

executors = {
    'default': ThreadPoolExecutor(max_workers=20),
}

job_defaults = {
    'coalesce': False,
    'max_instances': 1
}

scheduler = BackgroundScheduler(
    jobstores=jobstores,
    executors=executors,
    job_defaults=job_defaults,
    timezone='UTC'
)

def log_job_execution(task_name, data=None):
    timestamp = datetime.utcnow().isoformat()
    log_entry = {
        'timestamp': timestamp,
        'hostname': hostname,
        'task': task_name,
        'data': data
    }
    logger.info(f'Task executed: {json.dumps(log_entry)}')
    
    redis_conn.lpush('job_execution_log', json.dumps(log_entry))
    redis_conn.ltrim('job_execution_log', 0, 999)

def fetch_user_data():
    log_job_execution('fetch_user_data', {'users': 150})

def process_reports():
    log_job_execution('process_reports', {'reports': 42})

def cleanup_old_files():
    log_job_execution('cleanup_old_files', {'deleted': 237})

def send_daily_digest():
    log_job_execution('send_daily_digest', {'recipients': 1200})

def setup_distributed_jobs():
    scheduler.add_job(
        fetch_user_data,
        IntervalTrigger(minutes=5),
        id='fetch_users_' + hostname,
        name='Fetch User Data',
        replace_existing=True
    )
    
    scheduler.add_job(
        process_reports,
        CronTrigger(hour='0-23', minute='*/30'),
        id='process_reports_' + hostname,
        name='Process Reports Every 30 Minutes',
        replace_existing=True
    )
    
    scheduler.add_job(
        cleanup_old_files,
        CronTrigger(day_of_week='0', hour=3, minute=0),
        id='cleanup_files_' + hostname,
        name='Weekly Cleanup',
        replace_existing=True
    )
    
    scheduler.add_job(
        send_daily_digest,
        CronTrigger(hour=9, minute=0),
        id='daily_digest_' + hostname,
        name='Daily Digest Email',
        replace_existing=True
    )

def get_scheduler_info():
    jobs = scheduler.get_jobs()
    info = {
        'hostname': hostname,
        'running': scheduler.running,
        'job_count': len(jobs),
        'jobs': [
            {
                'id': job.id,
                'name': job.name,
                'next_run_time': str(job.next_run_time),
                'trigger': str(job.trigger)
            }
            for job in jobs
        ]
    }
    return info

if __name__ == '__main__':
    setup_distributed_jobs()
    scheduler.start()
    
    logger.info(f'Distributed scheduler started on {hostname}')
    logger.info(f'Scheduler info: {get_scheduler_info()}')
    
    try:
        while True:
            pass
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        logger.info('Scheduler shut down')

This version introduces several production-ready features. Each scheduler instance identifies itself by hostname, so you can run multiple instances and they’ll coordinate through Redis. Job IDs are unique per instance, preventing conflicts. The execution logs are stored in Redis as a list, which you can query to audit what jobs have run.

The key insight here is that Redis acts as a single source of truth. When you start multiple instances of this scheduler, they all read from the same Redis instance. Jobs are executed based on their schedule, but the coordination happens through Redis. This prevents duplicate executions across instances.

Monitoring and Debugging

Now let’s build a monitoring interface to see what’s happening in your scheduler. Create monitor.py:

import redis
import json
from datetime import datetime
import logging

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

class SchedulerMonitor:
    def __init__(self, redis_host='localhost', redis_port=6379, redis_db=0):
        self.redis = redis.Redis(
            host=redis_host,
            port=redis_port,
            db=redis_db,
            decode_responses=True
        )
    
    def get_all_jobs(self):
        jobs_raw = self.redis.get('apscheduler.jobs')
        if not jobs_raw:
            return []
        
        try:
            return json.loads(jobs_raw)
        except json.JSONDecodeError:
            return []
    
    def get_execution_history(self, limit=50):
        logs = self.redis.lrange('job_execution_log', 0, limit - 1)
        return [json.loads(log) for log in logs]
    
    def get_scheduler_stats(self):
        execution_logs = self.get_execution_history(1000)
        
        if not execution_logs:
            return {
                'total_executions': 0,
                'unique_tasks': 0,
                'by_hostname': {},
                'recent_errors': []
            }
        
        task_counts = {}
        hostname_counts = {}
        
        for log in execution_logs:
            task = log.get('task', 'unknown')
            hostname = log.get('hostname', 'unknown')
            
            task_counts[task] = task_counts.get(task, 0) + 1
            hostname_counts[hostname] = hostname_counts.get(hostname, 0) + 1
        
        return {
            'total_executions': len(execution_logs),
            'unique_tasks': len(task_counts),
            'executions_by_task': task_counts,
            'executions_by_hostname': hostname_counts,
            'last_execution': execution_logs[0] if execution_logs else None
        }
    
    def get_job_details(self, job_id):
        job_data = self.redis.get(f'apscheduler.job:{job_id}')
        if not job_data:
            return None
        return json.loads(job_data)
    
    def clear_execution_log(self):
        self.redis.delete('job_execution_log')
        logger.info('Execution log cleared')
    
    def disable_job(self, job_id):
        self.redis.hset('apscheduler.job_states', job_id, 'paused')
        logger.info(f'Job {job_id} disabled')
    
    def enable_job(self, job_id):
        self.redis.hset('apscheduler.job_states', job_id, 'active')
        logger.info(f'Job {job_id} enabled')

if __name__ == '__main__':
    monitor = SchedulerMonitor()
    
    print("=== Scheduler Statistics ===")
    stats = monitor.get_scheduler_stats()
    print(json.dumps(stats, indent=2))
    
    print("\n=== Recent Executions ===")
    history = monitor.get_execution_history(10)
    for entry in history:
        print(f"{entry['timestamp']} - {entry['hostname']}: {entry['task']}")

Run the monitor to see what your scheduler is doing:

python monitor.py

This gives you real-time visibility into job execution across all instances. The monitor pulls data directly from Redis, so it works whether you have one scheduler or fifty running in parallel.

For production environments, you might want to integrate this with logging systems. If you’re already using container orchestration and need centralized logging, check out how to stream container logs to Loki for comprehensive observability across your entire stack.

Now, let’s add error handling and retries. Create advanced_scheduler.py:

from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.jobstores.redis import RedisJobStore
from apscheduler.executors.pool import ThreadPoolExecutor
from apscheduler.triggers.cron import CronTrigger
from apscheduler.triggers.interval import IntervalTrigger
from apscheduler.events import EVENT_JOB_EXECUTED, EVENT_JOB_ERROR
import redis
import logging
import json
from datetime import datetime
from functools import wraps
import traceback

logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)

redis_conn = redis.Redis(host='localhost', port=6379, db=0, decode_responses=True)

jobstores = {
    'default': RedisJobStore(connection=redis_conn)
}

executors = {
    'default': ThreadPoolExecutor(max_workers=20),
}

job_defaults = {
    'coalesce': False,
    'max_instances': 1
}

scheduler = BackgroundScheduler(
    jobstores=jobstores,
    executors=executors,
    job_defaults=job_defaults,
    timezone='UTC'
)

def job_with_retry(max_retries=3, retry_delay_seconds=60):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = kwargs.pop('_retries', 0)
            try:
                result = func(*args, **kwargs)
                log_entry = {
                    'timestamp': datetime.utcnow().isoformat(),
                    'function': func.__name__,
                    'status': 'success',
                    'result': str(result)
                }
                redis_conn.lpush('job_results', json.dumps(log_entry))
                return result
            except Exception as e:
                error_msg = str(e)
                trace = traceback.format_exc()
                logger.error(f'Job {func.__name__} failed: {error_msg}')
                logger.error(f'Traceback: {trace}')
                
                log_entry = {
                    'timestamp': datetime.utcnow().isoformat(),
                    'function': func.__name__,
                    'status': 'error',
                    'error': error_msg,
                    'traceback': trace,
                    'retry_count': retries
                }
                redis_conn.lpush('job_errors', json.dumps(log_entry))
                
                if retries < max_retries:
                    logger.info(f'Scheduling retry {retries + 1}/{max_retries} for {func.__name__}')
                    scheduler.add_job(
                        wrapper,
                        'date',
                        run_date=datetime.utcnow().timestamp() + retry_delay_seconds,
                        args=args,
                        kwargs={**kwargs, '_retries': retries + 1},
                        id=f'{func.__name__}_retry_{retries + 1}_{datetime.utcnow().timestamp()}',
                        replace_existing=False
                    )
                else:
                    logger.error(f'Job {func.__name__} failed after {max_retries} retries')
                    log_entry['status'] = 'failed_permanently'
                    redis_conn.lpush('job_failed_permanently', json.dumps(log_entry))
        return wrapper
    return decorator

@job_with_retry(max_retries=3, retry_delay_seconds=30)
def unreliable_api_call():
    import random
    if random.random() < 0.7:
        raise Exception('API temporarily unavailable')
    logger.info('API call succeeded')
    return 'success'

@job_with_retry(max_retries=2, retry_delay_seconds=60)
def database_operation():
    logger.info('Database operation completed')
    return 'db_ok'

def job_error_listener(event):
    if event.exception:
        logger.error(f'Job {event.job_id} failed with exception: {event.exception}')

def job_success_listener(event):
    logger.info(f'Job {event.job_id} executed successfully')

scheduler.add_listener(job_error_listener, EVENT_JOB_ERROR)
scheduler.add_listener(job_success_listener, EVENT_JOB_EXECUTED)

def setup_jobs():
    scheduler.add_job(
        unreliable_api_call,
        IntervalTrigger(minutes=5),
        id='api_call_job',
        name='Unreliable API Call with Retries',
        replace_existing=True
    )
    
    scheduler.add_job(
        database_operation,
        CronTrigger(hour='*', minute=0),
        id='db_operation_job',
        name='Database Operation Every Hour',
        replace_existing=True
    )

if __name__ == '__main__':
    setup_jobs()
    scheduler.start()
    logger.info('Advanced scheduler started with retry capability')
    
    try:
        while True:
            pass
    except (KeyboardInterrupt, SystemExit):
        scheduler.shutdown()
        logger.info('Scheduler shut down')

This version adds automatic retry logic. If a job fails, it’s automatically rescheduled after a delay. All errors are logged to Redis for later analysis.

For API-heavy workloads, you’ll also want to think about authentication and security. If you’re calling external APIs from your scheduled jobs, take a look at securing microservice endpoints with OAuth2 bearer tokens to understand token management patterns that can be applied to your job credentials.

When jobs involve webhooks or external callbacks, security becomes critical. Understanding how to secure Telegram bot webhook endpoints provides patterns applicable to any webhook-triggered scheduler workflow.

Getting Started

Now you’re ready to deploy this to production. Here’s what I recommend:

For hosting your scheduler, Hetzner VPS offers excellent value and reliability. A 2-core instance with 4GB RAM handles thousands of scheduled jobs without issue. Alternatively, Contabo VPS provides similar performance at competitive pricing.

Your Redis instance can be on the same server as the scheduler for development, but in production, consider a dedicated Redis instance using managed services or a separate DigitalOcean droplet.

Start by testing locally with the examples I’ve provided. Deploy using systemd services to keep your scheduler running across reboots:

sudo nano /etc/systemd/system/apscheduler.service

[Unit]
Description=APScheduler Job Service
After=network.target redis.service

[Service]
Type=simple
User=www-data
WorkingDirectory=/opt/apscheduler
ExecStart=/opt/apscheduler/venv/bin/python distributed_scheduler.py
Restart=always
RestartSec=10

[Install]
WantedBy=multi-user.target

Enable and start it:

sudo systemctl enable apscheduler
sudo systemctl start apscheduler
sudo systemctl status apscheduler

Monitor logs with:

sudo journalctl -u apscheduler -f

If you’re orchestrating multiple services, n8n Cloud can be your central workflow hub, triggering and monitoring your APScheduler jobs via webhooks or the REST API.

Outsource Your Automation

Don’t have time? I build production n8n workflows, WhatsApp bots, and fully automated YouTube Shorts pipelines. Hire me on Fiverr, mention SYS3-HUGO for priority. Or DM at chasebot.online.

The combination of APScheduler and Redis is battle-tested and scalable. I’ve used this exact pattern for systems running millions of scheduled jobs daily. The beauty is in the simplicity: persistent state, distributed coordination, and straightforward monitoring. Your scheduled jobs are now production-grade.

Want to automate this yourself?

Start with n8n Cloud (free tier available) or self-host on a Hetzner VPS for full control.

Want this engine running on your own VPS?

This blog publishes itself — daily, unattended, on free API tiers. The full engine, Hugo theme, and setup guide are available as System 3.

Get System 3

Related Guides

system online