Task queue concept in Flask

Introduction

A task queue helps your Flask app do long jobs in the background. This keeps the app fast and responsive for users.

When sending emails after a user signs up without making them wait.

When processing uploaded files like images or videos after upload.

When running reports or data analysis that takes time.

When calling slow external services without blocking the user.

When scheduling tasks to run later or repeatedly.

Syntax

Flask

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from rq import Queue
from redis import Redis

app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
db = SQLAlchemy(app)

redis_connection = Redis()
task_queue = Queue('default', connection=redis_connection)

def background_task(n):
    """Example task that runs in the background."""
    print(f"Working on task {n}")
    return f"Task {n} done"

You need Redis installed and running for the queue to work.

Use rq library to create and manage task queues in Flask.

Examples

This adds a task to the queue to run background_task(5) later.

Flask

from rq import Queue
from redis import Redis

redis_connection = Redis()
task_queue = Queue('default', connection=redis_connection)

# Add a task to the queue
job = task_queue.enqueue(background_task, 5)
print(job.get_id())

Check if Redis is running before using the queue to avoid errors.

Flask

# What if Redis is not running?
try:
    redis_connection = Redis()
    redis_connection.ping()
except Exception as error:
    print("Redis server is not available.")

Handle the case when no tasks are waiting in the queue.

Flask

# What if the task queue is empty?
if task_queue.is_empty():
    print("No tasks to process right now.")

This schedules background_task(10) to run 10 minutes later.

Flask

# What if you want to schedule a task for later?
import datetime
from rq_scheduler import Scheduler

scheduler = Scheduler(connection=redis_connection)
scheduler.enqueue_at(datetime.datetime.utcnow() + datetime.timedelta(minutes=10), background_task, 10)

Sample Program

This Flask app has an endpoint /start-task that starts a background task to sleep for a given number of seconds. The task runs in the background, so the app stays responsive.

Flask

from flask import Flask, request, jsonify
from rq import Queue
from redis import Redis
import time

app = Flask(__name__)
redis_connection = Redis()
task_queue = Queue('default', connection=redis_connection)

def background_task(seconds):
    """Simulate a long task by sleeping."""
    print(f"Starting task for {seconds} seconds")
    time.sleep(seconds)
    print(f"Finished task for {seconds} seconds")
    return f"Slept for {seconds} seconds"

@app.route('/start-task', methods=['POST'])
def start_task():
    data = request.get_json() or {}
    seconds = data.get('seconds', 5)
    job = task_queue.enqueue(background_task, seconds)
    return jsonify({'message': 'Task started', 'job_id': job.get_id()})

if __name__ == '__main__':
    app.run(debug=True)

Important Notes

Task queues let your app handle slow jobs without making users wait.

Time complexity depends on the task itself, but queue operations are usually fast (O(1)).

Common mistake: forgetting to run a worker process that actually executes queued tasks.

Use task queues when you want to do work later or in parallel, not for instant responses.

Summary

Task queues run slow jobs in the background so your Flask app stays fast.

Redis and rq are popular tools to create task queues in Flask.

Always run a worker to process tasks from the queue.