Flask-Threads

0.2.0 · active · verified Sat Apr 11

Flask-Threads is a helper library designed to simplify working with threads within Flask applications. It addresses the common challenge of maintaining the Flask application context (e.g., `flask.g`, `request`) when executing code in background threads or using concurrent futures, which are typically thread-local. The library ensures that thread-local proxies remain accessible, preventing `RuntimeError` exceptions that occur when trying to access context outside the main request thread. The current version is 0.2.0, released on May 20, 2025, with an infrequent release cadence, primarily focusing on Flask compatibility.

Warnings

breaking Older versions of Flask-Threads might not be compatible with Flask versions 3.0.0 and above.
Fix: Upgrade to Flask-Threads version 0.2.0 or newer: `pip install --upgrade Flask-Threads`. Ensure your Flask version is compatible.
gotcha When running Flask in development mode with `debug=True`, Flask's reloader often starts the application twice. This can lead to background threads (including those managed by Flask-Threads) being initialized and run twice, causing unexpected behavior.
Fix: To prevent this in development, run your Flask application with `app.run(debug=True, use_reloader=False)`. For production, use a WSGI server like Gunicorn or uWSGI, which manage processes differently and typically don't have this issue.
gotcha While Flask-Threads helps maintain application context in background threads, it's crucial to understand that Flask's `request`, `g`, and `session` proxies are fundamentally tied to a specific request's lifecycle and thread. Misusing them (e.g., trying to modify `request` state in a background thread or keeping the context alive unnecessarily long) can still lead to data leaks, errors, or unexpected behavior.
Fix: Only access context-local objects in background threads for read-only purposes or when `flask-threads` explicitly manages the context copy. For complex, long-running background jobs, consider external task queues like Celery or RQ, and pass only plain, serializable data (like IDs or file paths) rather than Flask context objects.
gotcha Python's Global Interpreter Lock (GIL) limits true CPU parallelism for threads. While `flask-threads` is excellent for I/O-bound tasks that need Flask context, it will not make CPU-bound tasks run faster by simply using more threads.
Fix: For CPU-bound tasks, consider using multi-processing (e.g., Python's `multiprocessing` module or `concurrent.futures.ProcessPoolExecutor`), or dedicated worker queues, which can leverage multiple CPU cores.

Install

pip install Flask-Threads Install with pip

Imports

AppContextThread

from flaskthreads import AppContextThread

ThreadPoolWithAppContextExecutor

from flaskthreads import ThreadPoolWithAppContextExecutor

Quickstart

This example demonstrates how to use `AppContextThread` and `ThreadPoolWithAppContextExecutor` to run background tasks while retaining access to Flask's application context, specifically `flask.g`. The `user-id` is set in `flask.g` within the main request thread and then accessed correctly by the function running in a separate thread.

from flask import g, request, Flask
from flaskthreads import AppContextThread, ThreadPoolWithAppContextExecutor
import time

app = Flask('my_app')

def do_some_user_work_in_another_thread():
    # Accessing flask.g from a different thread, enabled by Flask-Threads
    user_id = g.user_id
    print(f"[Thread] User ID from g: {user_id}")
    time.sleep(1) # Simulate work
    return f"Processed user {user_id}"

@app.route('/user/thread')
def get_user_with_thread():
    g.user_id = request.headers.get('user-id', 'default_user_id_thread')
    print(f"[Main] Setting g.user_id: {g.user_id}")

    t = AppContextThread(target=do_some_user_work_in_another_thread)
    t.start()
    t.join() # Wait for the thread to complete
    return 'OK via AppContextThread'

@app.route('/user/executor')
def get_user_with_executor():
    g.user_id = request.headers.get('user-id', 'default_user_id_executor')
    print(f"[Main] Setting g.user_id: {g.user_id}")

    with ThreadPoolWithAppContextExecutor(max_workers=2) as pool:
        future = pool.submit(do_some_user_work_in_another_thread)
        result = future.result() # Wait for the future to complete
        print(f"[Main] Future result: {result}")
    return 'OK via ThreadPoolWithAppContextExecutor'

if __name__ == '__main__':
    # For demonstration, use a simple run. In production, use a WSGI server.
    # Set 'user-id' header (e.g., with curl -H 'user-id: 123' http://127.0.0.1:5000/user/thread)
    app.run(debug=True, use_reloader=False) # use_reloader=False to avoid double thread start in dev

view raw JSON →