Scheduled Tasks in Python
Have you ever needed a Python script to run automatically at a specific time? Maybe you want to send a daily report every morning, back up your database every night, or check for new data every hour. Instead of manually running your script each time, you can set up scheduled tasks that run themselves.
In this tutorial, you’ll learn several ways to schedule Python scripts to run automatically. We’ll cover tools for beginners and production systems. By the end, you’ll know which approach fits your needs.
What is Cron?
Cron is a time-based job scheduler that comes pre-installed on most Unix and Linux systems (including macOS). It allows you to specify when commands should run using a “cron expression.”
Think of cron like an alarm clock for your computer. You set it once, and it reminds you (runs your script) at the specified times automatically.
A cron expression consists of five fields:
┌───────────── minute (0 - 59)
│ ┌───────────── hour (0 - 23)
│ │ ┌───────────── day of month (1 - 31)
│ │ │ ┌───────────── month (1 - 12)
│ │ │ │ ┌───────────── day of week (0 - 6, Sunday = 0)
│ │ │ │ │
* * * * *
Each asterisk means “any value.” Here are some common examples:
# Run at 9:00 AM every day
0 9 * * *
# Run every hour (at minute 0)
0 * * * *
# Run every Monday at 6:00 AM
0 6 * * 1
# Run every 15 minutes
*/15 * * * *
Running Python Scripts with Cron
To run a Python script with cron, you need to know the full path to both Python and your script. Let’s create a simple example.
First, create a script called daily_report.py (see the datetime module for the timestamp formatting used here):
#!/usr/bin/env python3
"""Daily report generator that runs automatically."""
from datetime import datetime
from pathlib import Path
# Path to the output file
output_file = Path("/home/user/reports/daily.txt")
def generate_report():
"""Create a simple daily report."""
now = datetime.now()
report = f"Report generated: {now.strftime('%Y-%m-%d %H:%M:%S')}\n"
report += "Tasks completed: 42\n"
report += "Errors: 0\n"
# Write the report
output_file.parent.mkdir(parents=True, exist_ok=True)
output_file.write_text(report)
print(f"Report written to {output_file}")
if __name__ == "__main__":
generate_report()
Now add this script to your crontab:
# Open the crontab editor
crontab -e
# Add this line to run the script every day at 9:00 AM
# Redirect output to a log file so you can debug failures
0 9 * * * /usr/bin/python3 /home/user/scripts/daily_report.py >> /var/log/daily_report.log 2>&1
The crontab editor will open. Add your cron line at the end, save, and exit. After saving and exiting, your script will run automatically every morning at 9 AM.
To see your current crontab:
crontab -l
To remove your cron jobs:
crontab -r
Using Virtual Environments in Cron
Most Python projects use virtual environments. Running /usr/bin/python3 won’t use your project’s dependencies. Use the path to your venv’s Python:
# Run with virtual environment Python
0 9 * * * /home/user/project/venv/bin/python /home/user/scripts/daily_report.py >> /var/log/daily_report.log 2>&1
Handling Timezones
Cron uses the system timezone, which often surprises developers. A job set to run at “9 AM” will run at 9 AM in whatever timezone the system is configured for (often UTC on servers).
You can override this for a specific job by setting the TZ environment variable:
# Run at 9 AM London time
0 9 * * * TZ=Europe/London /usr/bin/python3 /home/user/scripts/daily_report.py >> /var/log/daily_report.log 2>&1
Preventing Overlapping Runs
If a job takes longer than expected, the next scheduled run might start before the previous one finishes, causing data issues. Use a lock file to prevent this:
#!/usr/bin/env python3
"""Daily report with lock to prevent overlapping runs."""
import fcntl
import sys
import time
from pathlib import Path
lock_file = Path("/tmp/daily_report.lock")
def run_with_lock():
"""Acquire exclusive lock before running."""
with open(lock_file, 'w') as f:
try:
fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
except BlockingIOError:
print("Another instance is already running. Exiting.")
sys.exit(0)
# Run your actual task here
print("Running daily report...")
time.sleep(5) # Simulate work
print("Done.")
if __name__ == "__main__":
run_with_lock()
Install the filelock package for a simpler API:
pip install filelock
from filelock import FileLock
lock = FileLock("/tmp/daily_report.lock")
with lock:
# Your task runs here
print("Running daily report...")
The python-crontab Module
The python-crontab package lets you create and modify cron jobs directly from Python code. This is useful when your application needs to dynamically schedule tasks.
Install it first:
pip install python-crontab
Here’s how to create a cron job programmatically:
from crontab import CronTab
# Create a cron job for the current user
cron = CronTab(user=True)
# Create a new job
job = cron.new(
command='/usr/bin/python3 /home/user/scripts/daily_report.py >> /home/user/logs/daily_report.log 2>&1',
comment='daily_report_task'
)
# Set it to run at 9:00 AM every day
job.setall('0 9 * * *')
# Write the cron jobs to the system
cron.write()
print("Cron job created successfully!")
# output: Cron job created successfully!
You can also list, modify, and remove cron jobs:
from crontab import CronTab
cron = CronTab(user=True)
# List all jobs (use as_string=True for readable output)
for job in cron:
print(f"Command: {job.command}, Schedule: {job.schedule(as_string=True)}")
# Find and remove a specific job
for job in cron:
if 'daily_report.py' in job.command:
job.delete()
# Write once after removing
cron.write()
print("Job removed")
This approach is helpful when building admin interfaces or tools that let users customize their scheduled tasks.
The schedule Module
The schedule library runs scheduled tasks from within your Python application. It works well for scripts that run continuously.
Install it:
pip install schedule
Here’s a basic example:
import schedule
import time
from datetime import datetime
def job():
"""A simple job that prints the current time."""
print(f"Job ran at {datetime.now()}")
# Schedule jobs using a clean Python syntax
schedule.every().day.at("09:00").do(job)
schedule.every().hour.do(job)
schedule.every(30).minutes.do(job)
print("Scheduler started, press Ctrl+C to stop")
# Run the scheduler loop
while True:
schedule.run_pending()
time.sleep(1)
This script prints “Job ran at” every hour and every 30 minutes, plus every day at 9:00 AM. The run_pending() function checks if any scheduled jobs are due to run.
You can also pass arguments to your scheduled functions:
def send_report(user_id):
print(f"Sending report for user {user_id}")
# Schedule with arguments
schedule.every().day.at("09:00").do(send_report, user_id=42)
The schedule module is great for simple use cases but has a limitation: it only runs while your Python script is actively executing. If your script stops, no jobs will run.
APScheduler
APScheduler is a full-featured scheduler with background execution, job persistence, and complex trigger support.
Install APScheduler:
pip install apscheduler
Here’s a simple example using the blocking scheduler:
from apscheduler.schedulers.blocking import BlockingScheduler
from datetime import datetime
def my_job():
"""A job that prints the current time."""
print(f"Job executed at {datetime.now()}")
# Create the scheduler
scheduler = BlockingScheduler()
# Add a job that runs every hour
scheduler.add_job(
my_job,
'interval',
hours=1,
id='my_job_id'
)
# Add a job that runs at a specific time
scheduler.add_job(
my_job,
'cron',
hour=9,
minute=0,
id='daily_job'
)
print("Scheduler starting...")
try:
scheduler.start()
except (KeyboardInterrupt, SystemExit):
print("Scheduler stopped")
APScheduler supports multiple trigger types:
interval: Run every N seconds, minutes, hours, or daysdate: Run once at a specific date and timecron: Run at specific times using cron-like syntax
You can also use APScheduler with different schedulers:
from apscheduler.schedulers.background import BackgroundScheduler
from datetime import datetime
def my_job():
"""A job that prints the current time."""
print(f"Job executed at {datetime.now()}")
# Background scheduler doesn't block the main thread
scheduler = BackgroundScheduler()
scheduler.add_job(my_job, 'interval', minutes=30)
scheduler.start()
# Your main code continues here
print("Main program continuing...")
APScheduler also supports job stores that persist jobs to databases, which is useful when you need jobs to survive application restarts.
Celery Beat
Celery Beat is part of the Celery distributed task queue system. It handles scheduling for applications that need to distribute work across multiple machines or handle heavy workloads.
Install Celery with the beat scheduler:
pip install celery[redis]
Here’s a basic setup. First, create a file called tasks.py:
from celery import Celery
from celery.schedules import crontab
# Configure Celery with Redis as the broker
app = Celery('myapp', broker='redis://localhost:6379/0')
@app.task
def generate_report():
"""Generate a daily report."""
print("Generating report...")
return "Report generated"
# Configure beat schedule
app.conf.beat_schedule = {
'daily-report': {
'task': 'tasks.generate_report',
'schedule': crontab(hour=9, minute=0),
},
'hourly-check': {
'task': 'tasks.generate_report',
'schedule': 3600, # Every hour in seconds
},
}
Run the Celery worker:
celery -A tasks worker --loglevel=info
Run the beat scheduler in a separate terminal:
celery -A tasks beat
This setup requires Redis to be running. Celery Beat is more complex but handles large-scale task distribution well.
Systemd Timers
If you’re on Linux, your system probably uses systemd. Systemd timers are the modern replacement for cron on many Linux distributions. They integrate directly with systemd and can trigger on system events.
Create a service file at /home/user/.config/systemd/user/daily-report.service:
[Unit]
Description=Run daily report script
[Service]
Type=oneshot
ExecStart=/home/user/project/venv/bin/python /home/user/scripts/daily_report.py
Create a timer file at /home/user/.config/systemd/user/daily-report.timer:
[Unit]
Description=Run daily report every day at 9 AM
[Timer]
OnCalendar=*-*-* 09:00:00
Persistent=true
[Install]
WantedBy=timers.target
Enable and start the timer:
# Reload systemd to pick up the new files
systemctl --user daemon-reload
# Start the timer
systemctl --user start daily-report.timer
# Enable it to run on boot
systemctl --user enable daily-report.timer
# Check status
systemctl --user list-timers
Systemd timers offer features like:
- Run tasks when the system is idle
- Retry failed jobs automatically
- Track job execution history with journalctl
This approach works well for system-level automation on Linux servers.
Which Approach Should You Use?
Here’s a quick guide to choose the right scheduling method:
- schedule module: Quick scripts, prototypes, simple needs
- python-crontab: Manage cron jobs from Python code
- APScheduler: More complex scheduling within a Python app
- Celery Beat: Distributed systems, multiple workers
- Systemd timers: Linux system automation, root-level tasks
For most beginners, start with the schedule module or simple cron. As your needs grow, move to APScheduler or Celery.
Conclusion
You now have several tools to schedule Python scripts automatically. Choose the tool that matches your current requirements. As you build more complex applications, you’ll have the tools to scale up.
Always test your script manually before automating it, and keep logs to help debug failures.
The right scheduling tool depends on your specific needs. Don’t overcomplicate things at the start. You can always upgrade to a more powerful solution later.
See Also
- Logging in Python — centralize output from scheduled scripts
- Classes and Objects — structuring larger scripts that run on a schedule
- Working with Dates and Times — more on the
datetimemodule used in the examples above