Practice Now
Celery
Background Tasks
Python
Error Handling
Flask

Working outside of request context error with Celery background task

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design
Kafka & StreamingPythonDebugging & Errors

Introduction

The "Working outside of request context" error in Flask + Celery occurs when a Celery task tries to access request, session, g, or other request-scoped objects. Celery tasks run in a separate worker process with no HTTP request, so Flask's request context does not exist. The fix is to pass all needed data as task arguments instead of accessing the request inside the task. If you need current_app (for config or extensions), use app.app_context() to create an application context inside the task.

The Error

python
1from flask import request
2from celery import Celery
3
4celery = Celery('tasks', broker='redis://localhost:6379')
5
6@celery.task
7def process_upload(filename):
8    # This fails — no request context in Celery worker
9    user_id = request.headers.get('X-User-Id')
10    # RuntimeError: Working outside of request context.

Extract all needed data from the request in the Flask route and pass it to the task:

python
1from flask import Flask, request, jsonify
2from celery import Celery
3
4app = Flask(__name__)
5celery = Celery('tasks', broker='redis://localhost:6379')
6
7@celery.task
8def process_upload(filename, user_id, content_type):
9    # All data is passed as arguments — no request needed
10    print(f"Processing {filename} for user {user_id}")
11    # Do the actual work...
12    return {"status": "completed", "filename": filename}
13
14@app.route('/upload', methods=['POST'])
15def upload():
16    file = request.files['file']
17    user_id = request.headers.get('X-User-Id')
18    content_type = request.content_type
19
20    # Extract data from request, pass to task
21    task = process_upload.delay(file.filename, user_id, content_type)
22    return jsonify({"task_id": task.id}), 202

Fix 2: Application Context for current_app

If the task needs Flask app configuration or extensions (database, mail):

python
1from flask import Flask
2from celery import Celery
3
4app = Flask(__name__)
5app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///app.db'
6celery = Celery('tasks', broker='redis://localhost:6379')
7
8# Configure Celery to use Flask app context
9class FlaskTask(celery.Task):
10    def __call__(self, *args, **kwargs):
11        with app.app_context():
12            return self.run(*args, **kwargs)
13
14celery.Task = FlaskTask
15
16@celery.task
17def send_welcome_email(user_id):
18    # app_context is active — can use current_app, db, mail, etc.
19    from flask import current_app
20    user = User.query.get(user_id)
21    mail_server = current_app.config['MAIL_SERVER']
22    # Send email...

Fix 3: Flask Factory Pattern

For applications using the factory pattern:

python
1# app/__init__.py
2from flask import Flask
3from celery import Celery
4
5celery = Celery(__name__)
6
7def create_app():
8    app = Flask(__name__)
9    app.config.from_object('config')
10
11    # Configure Celery
12    celery.conf.update(app.config)
13
14    class ContextTask(celery.Task):
15        def __call__(self, *args, **kwargs):
16            with app.app_context():
17                return self.run(*args, **kwargs)
18
19    celery.Task = ContextTask
20
21    # Register blueprints, extensions, etc.
22    from app.extensions import db, mail
23    db.init_app(app)
24    mail.init_app(app)
25
26    return app
python
1# app/tasks.py
2from app import celery
3from app.extensions import db
4from app.models import User
5
6@celery.task
7def process_user(user_id):
8    # App context is provided by ContextTask
9    user = User.query.get(user_id)
10    user.processed = True
11    db.session.commit()

Fix 4: Manual App Context

python
1@celery.task
2def generate_report(report_params):
3    # Manually push app context
4    from app import create_app
5    app = create_app()
6
7    with app.app_context():
8        from app.models import Order
9        orders = Order.query.filter_by(**report_params).all()
10        # Generate report...
11        return {"count": len(orders)}

This is simpler but creates a new app instance per task, which is less efficient.

What You Cannot Do in a Celery Task

python
1# NONE of these work in Celery — no request context
2from flask import request, session, g
3
4@celery.task
5def bad_task():
6    request.args.get('key')      # RuntimeError
7    session['user_id']            # RuntimeError
8    g.db                          # RuntimeError
9    request.headers['Auth']       # RuntimeError

Task Status Checking

python
1@app.route('/upload', methods=['POST'])
2def upload():
3    task = process_upload.delay(request.files['file'].filename)
4    return jsonify({"task_id": task.id}), 202
5
6@app.route('/status/<task_id>')
7def task_status(task_id):
8    task = process_upload.AsyncResult(task_id)
9    return jsonify({
10        "task_id": task.id,
11        "status": task.status,
12        "result": task.result if task.ready() else None
13    })

Common Pitfalls

  • Accessing request, session, or g inside a Celery task: Celery tasks run in a separate worker process with no HTTP request. There is no request context. Always pass needed data as function arguments from the Flask route handler.
  • Confusing application context with request context: app.app_context() provides access to current_app and app-level extensions (SQLAlchemy, Mail), but NOT to request or session. These are separate contexts. You can create an app context in a task, but never a request context.
  • Passing non-serializable objects as task arguments: Celery serializes task arguments (default JSON). Passing Flask request objects, file handles, or SQLAlchemy model instances fails. Pass primitive types (strings, ints, dicts) and re-query the database inside the task.
  • Not running the Celery worker with the Flask app context: If you skip the ContextTask base class, tasks that use db.session or current_app.config fail. Always configure Celery to wrap tasks in app.app_context().
  • Creating a new Flask app instance in every task call: While app = create_app() inside a task works, it creates a new app and re-initializes all extensions on every task execution. Use the ContextTask base class pattern to reuse a single app instance across all tasks in the worker.

Summary

  • Pass all request data (user ID, headers, form data) as task arguments — never access request inside a task
  • Use a custom ContextTask base class to wrap tasks in app.app_context() for database and config access
  • Celery tasks can use current_app (with app context) but never request or session
  • Serialize only primitive types as task arguments — re-query models inside the task
  • Use the Flask factory pattern with celery.Task = ContextTask for production applications
Related reading
Course
Beginner
27 lessons
10 hours
System Design Fundamentals

Build a strong foundation in designing scalable, reliable distributed systems.

View the course
Track what you have practised

A free account saves your progress, solutions and study plan across every problem on Codemia.

System Design practice on Codemia

Work through 120+ system design problems with detailed solutions, from rate limiters to multi-region storage.

Practice system design
Codemia
© 2026 Codemia
Resources
Legal
Social
All Rights Reserved.