Flask Scheduled Tasks
In web applications, certain operations need to run on a schedule rather than in response to user requests. These might include database cleanup, sending daily emails, generating reports, or updating cached data. This guide will show you how to implement scheduled tasks in your Flask application.
Introduction to Scheduled Tasks
Scheduled tasks (or cron jobs in Unix/Linux terminology) are processes that run automatically at predefined intervals or at specific times. In a Flask application, scheduled tasks allow you to:
- Perform regular maintenance operations
- Update data periodically
- Send scheduled notifications or reports
- Process data in the background at predetermined times
Let's explore different approaches to implementing scheduled tasks in Flask applications.
Approaches to Scheduling Tasks in Flask
1. Flask-APScheduler
Flask-APScheduler is an extension for Flask that adds support for the APScheduler library, making it easy to schedule tasks within your Flask application.
Installation
pip install Flask-APScheduler
Basic Implementation
Here's a simple example of using Flask-APScheduler to run a task every minute:
from flask import Flask
from flask_apscheduler import APScheduler
app = Flask(__name__)
scheduler = APScheduler()
# Define a task to be scheduled
@scheduler.task('interval', id='my_job', seconds=60)
def my_scheduled_job():
print('This job runs every minute')
# Your task logic here
@app.route('/')
def home():
return "Flask Scheduler App"
if __name__ == '__main__':
scheduler.init_app(app)
scheduler.start()
app.run(debug=True)
The output will show "This job runs every minute" in your console every 60 seconds while the application is running.
Job Types in APScheduler
APScheduler supports different types of schedules:
-
Interval Jobs - Run periodically at fixed intervals
python@scheduler.task('interval', id='interval_task', hours=24)
def daily_task():
print('This task runs every 24 hours') -
Cron Jobs - Run at specific times, dates, or intervals using cron expressions
python# Run at 10:30 AM every weekday (Monday through Friday)
@scheduler.task('cron', id='cron_task', day_of_week='mon-fri', hour=10, minute=30)
def weekday_morning_task():
print('This task runs at 10:30 AM every weekday') -
Date Jobs - Run once at a specific date and time
pythonfrom datetime import datetime
import pytz
# Run once at the specified datetime
target_date = datetime(2023, 12, 31, 23, 59, 59, tzinfo=pytz.UTC)
@scheduler.task('date', id='date_task', run_date=target_date)
def new_year_task():
print('Happy New Year! This runs once at the specified time')
2. Celery with Celery Beat
For more complex scheduling needs, especially in production environments, Celery with Celery Beat provides a robust solution.
Installation
pip install celery flask
Basic Implementation
First, create a Flask application with Celery integration:
# app.py
from flask import Flask
from celery import Celery
def create_app():
app = Flask(__name__)
app.config.update(
CELERY_BROKER_URL='redis://localhost:6379/0',
CELERY_RESULT_BACKEND='redis://localhost:6379/0'
)
return app
def create_celery(app):
celery = Celery(
app.import_name,
backend=app.config['CELERY_RESULT_BACKEND'],
broker=app.config['CELERY_BROKER_URL']
)
celery.conf.update(app.config)
class ContextTask(celery.Task):
def __call__(self, *args, **kwargs):
with app.app_context():
return self.run(*args, **kwargs)
celery.Task = ContextTask
return celery
app = create_app()
celery = create_celery(app)
@app.route('/')
def home():
return "Flask with Celery Scheduler"
@celery.task()
def scheduled_task():
print('This is a scheduled task with Celery Beat')
# Your task logic here
return "Task completed"
Then, create a schedule configuration file:
# celerybeat_config.py
from celery.schedules import crontab
CELERYBEAT_SCHEDULE = {
'run-every-morning': {
'task': 'app.scheduled_task',
'schedule': crontab(hour=7, minute=0), # Runs daily at 7:00 AM
},
'run-every-fifteen-minutes': {
'task': 'app.scheduled_task',
'schedule': 15 * 60, # 15 minutes in seconds
},
}
To run the scheduler:
- Start Redis (or another message broker)
- Start a Celery worker:
celery -A app.celery worker --loglevel=info - Start Celery Beat:
celery -A app.celery beat --loglevel=info --schedule=/tmp/celerybeat-schedule - Start your Flask application:
python app.py
3. Simple Background Thread (for development only)
For smaller applications or during development, you might use a simple background thread:
from flask import Flask
import threading
import time
app = Flask(__name__)
def background_task():
while True:
print('Running background task...')
# Your task logic here
time.sleep(60) # Wait for 60 seconds
@app.route('/')
def home():
return "Flask with Background Thread"
if __name__ == '__main__':
# Start the background task in a separate thread
thread = threading.Thread(target=background_task)
thread.daemon = True # This ensures the thread will exit when the main program exits
thread.start()
app.run(debug=True)
⚠️ Note: This approach is only recommended for development or very simple applications. For production use cases, prefer Flask-APScheduler or Celery.
Real-World Example: Daily Report Generator
Let's build a more complete example - a Flask application that generates and emails daily reports:
from flask import Flask, render_template
from flask_apscheduler import APScheduler
import pandas as pd
from datetime import datetime
import smtplib
from email.mime.multipart import MIMEMultipart
from email.mime.text import MIMEText
import os
app = Flask(__name__)
scheduler = APScheduler()
# Simulating database access
def get_daily_stats():
# In a real app, this would query your database
# For this example, we'll generate some dummy data
return {
'total_users': 1250,
'new_users': 34,
'active_users': 876,
'revenue': 12500.75
}
def generate_report():
stats = get_daily_stats()
report_date = datetime.now().strftime('%Y-%m-%d')
# Create a simple report
df = pd.DataFrame([stats])
report_path = f'reports/daily_report_{report_date}.csv'
# Ensure directory exists
os.makedirs('reports', exist_ok=True)
# Save report to CSV
df.to_csv(report_path, index=False)
print(f"Report generated: {report_path}")
return report_path
def send_email_report(report_path):
# Email configuration
sender_email = '[email protected]'
receiver_email = '[email protected]'
password = 'your_email_password' # Use environment variables in production
# Create message
msg = MIMEMultipart()
msg['From'] = sender_email
msg['To'] = receiver_email
msg['Subject'] = f'Daily Report - {datetime.now().strftime("%Y-%m-%d")}'
# Attach body
body = "Please find attached the daily report."
msg.attach(MIMEText(body, 'plain'))
# Read the report file
with open(report_path, 'r') as file:
report_content = file.read()
# Attach report as plain text (in a real scenario, you might attach the file)
report_attachment = MIMEText(report_content)
report_attachment.add_header('Content-Disposition', f'attachment; filename="{os.path.basename(report_path)}"')
msg.attach(report_attachment)
try:
# Connect to server and send email
server = smtplib.SMTP('smtp.gmail.com', 587)
server.starttls()
server.login(sender_email, password)
server.send_message(msg)
server.quit()
print("Email sent successfully")
except Exception as e:
print(f"Failed to send email: {e}")
# Schedule the daily report task
@scheduler.task('cron', id='daily_report', hour=6, minute=0)
def scheduled_daily_report():
print(f"Generating daily report at {datetime.now()}")
report_path = generate_report()
send_email_report(report_path)
@app.route('/')
def home():
return render_template('home.html', title="Daily Report Generator")
@app.route('/generate-now')
def generate_now():
# Allow manual triggering of the report
report_path = generate_report()
send_email_report(report_path)
return "Report generated and sent!"
if __name__ == '__main__':
scheduler.init_app(app)
scheduler.start()
app.run(debug=True)
In this example, we've created a Flask application that:
- Generates a daily report at 6:00 AM using APScheduler
- Allows manual generation of reports via a web endpoint
- Simulates gathering statistics from a database
- Saves reports as CSV files
- Emails the reports to a specified recipient
Best Practices for Scheduled Tasks
- Error Handling: Always implement robust error handling in scheduled tasks, as there won't be a user to see error messages.
@scheduler.task('interval', id='error_handling_example', minutes=30)
def task_with_error_handling():
try:
# Your task logic here
result = perform_some_operation()
print(f"Task completed successfully: {result}")
except Exception as e:
# Log the error
print(f"Error in scheduled task: {e}")
# You might want to send an alert or notification here
- Logging: Implement proper logging instead of print statements.
import logging
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.FileHandler("scheduler.log"),
logging.StreamHandler()
]
)
@scheduler.task('interval', id='logging_example', hours=1)
def task_with_logging():
logging.info("Starting scheduled task")
try:
# Task logic
result = complex_operation()
logging.info(f"Task completed with result: {result}")
except Exception as e:
logging.error(f"Task failed: {e}", exc_info=True)
- Database Connections: Be careful with database connections in long-running tasks.
@scheduler.task('interval', id='db_example', hours=4)
def database_cleanup_task():
from flask_sqlalchemy import SQLAlchemy
from app import create_app
app = create_app()
db = SQLAlchemy(app)
with app.app_context():
# Perform database operations
old_records = OldRecords.query.filter(OldRecords.date < cutoff_date).all()
for record in old_records:
db.session.delete(record)
db.session.commit()
-
Task Duration: Be mindful of how long your tasks run to avoid overlapping executions.
-
Production Deployment: For production, use separate processes for your web application and scheduled tasks.
Summary
Scheduled tasks are essential for automating routine operations in Flask applications. We've covered:
- Using Flask-APScheduler for simpler applications
- Implementing Celery with Celery Beat for more complex scheduling needs
- Creating scheduled tasks with different timing patterns (interval, cron, one-time)
- Building a real-world example of a scheduled reporting system
- Best practices for implementing scheduled tasks
By choosing the right approach and following best practices, you can create reliable scheduled tasks that enhance your Flask application's functionality without requiring manual intervention.
Additional Resources
- Flask-APScheduler Documentation
- APScheduler Documentation
- Celery Documentation
- Crontab Guru - A quick reference for cron schedule expressions
Exercises
- Create a Flask application with a scheduled task that checks a website's availability every 5 minutes and logs the results.
- Implement a daily database backup task using Flask-APScheduler.
- Build a Flask app with Celery Beat that generates and emails a weekly report summarizing user activity.
- Extend the daily report generator example to include charts and graphs in the email using a library like matplotlib.
- Create a scheduled task that periodically cleans up old files from a specific directory.
If you spot any mistakes on this website, please let me know at [email protected]. I’d greatly appreciate your feedback! :)