Analysis and solution for request context problems in Flask background threads

introduction

In Flask development, we often encounter threads in the background (such asorceleryState of time-consuming operations performed in tasks). However, if you access Flask's directly in the background threadrequestThe object will meetRuntimeError: Working outside of request contextmistake.

This article will use a real case to analyze the causes of errors and provide 3 solutions to help you correctly handle request contexts in Flask background tasks.

Problem background

Error log analysis

In the log, we found the following error:

2025-05-15 23:20:08,759 - app - ERROR - There was an error in processing: Failed to save the operation log
Traceback (most recent call last):
  File "/doudian-phone-tool/services/order_service.py", line 129, in save_operation_log
    auth_token, user_id = PassportService.current_user_id()
                          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/doudian-phone-tool/libs/", line 33, in current_user_id
    auth_header = ("Authorization", "")
                  ^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/werkzeug/", line 311, in __get__
    obj = instance._get_current_object()
          ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "/usr/local/lib/python3.11/site-packages/werkzeug/", line 508, in _get_current_object
    raise RuntimeError(unbound_message) from None
RuntimeError: Working outside of request context.

Cause of error:

Called in the background threadPassportService.current_user_id(),andcurrent_user_id()Dependence(Flask request context).
Because the background thread does not have Flask's request context, it leads toRuntimeError。

Solution

Method 1: Get user_id in advance and pass it into the background thread (recommended)

Core idea

Get it in the main thread (HTTP request context)user_id, and then pass it to the background thread to avoid direct access of the background threadrequest。

Code implementation

def process_file_background(user_id):  # Receive user_id parameter    """Background threads process files"""
    from app import app
    with app.app_context():
        try:
            output_file = process_and_export_results(
                raw_results=raw_results,
                filepath=filepath,
                original_filename=original_filename,
                cookie=cookie,
                nationwide=nationwide,
                userId=user_id,  # Use the incoming user_id directly                receiver_email=receiver_email
            )
            (f"File processing is completed: {output_file}")
            if (filepath):
                (filepath)
        except Exception as e:
            (f"Background file processing failed: {str(e)}", exc_info=True)

# Get user_id in the main thread and pass it to the background threadauth_token, user_id = PassportService.current_user_id()
thread = (target=process_file_background, args=(user_id,))  # Pass in user_id()

advantage

Avoid background thread access completelyrequest
Clear code logic, easy to maintain

Method 2: Handle situations without request context in save_operation_log

Core idea

If logging does not require strong dependenciesuser_id, can be modifiedsave_operation_log, causing it to skip or use the default value without a request context.

Code implementation

@staticmethod
def save_operation_log(
        business_type: str = 'Upload',
        operation_type: str = 'Start upload',
        operation_content: str = None,
        operation_params: str = None,
        user_id: int = None,  # Add optional parameters):
    """Save the operation log"""
    try:
        from app import app
        with app.app_context():
            if user_id is None:  # If no user_id is passed in, try to get it                try:
                    auth_token, user_id = PassportService.current_user_id()
                except RuntimeError:  # If not in the request context, record anonymous logs                    user_id = 0  # Or use None, depending on business needs            memberOperationLog = MemberOperationLog(
                user_id=user_id,
                business_type=business_type,
                operation_type=operation_type,
                operation_content=operation_content,
                operation_params=operation_params,
                operation_time=(),
                create_time=(),
                update_time=()
            )
            (memberOperationLog)
            ()
    except Exception as e:
        raise MemberOperationLogError("Save the operation log failed")

Applicable scenarios

Logging is not mandatoryuser_id
Allow some logs to have no user information

Method 3: Use Flask's copy_current_request_context (suitable for simple tasks)

Core idea

Using the provided by Flaskcopy_current_request_contextDecorator, copy the request context to the background thread.

Code implementation

from flask import copy_current_request_context

def process_file_background():
    """Background threads process files (carrying request context)"""
    @copy_current_request_context  # Copy the request context    def run_in_context():
        from app import app
        with app.app_context():
            try:
                output_file = process_and_export_results(
                    raw_results=raw_results,
                    filepath=filepath,
                    original_filename=original_filename,
                    cookie=cookie,
                    nationwide=nationwide,
                    userId=user_id,
                    receiver_email=receiver_email
                )
                (f"File processing is completed: {output_file}")
                if (filepath):
                    (filepath)
            except Exception as e:
                (f"Background file processing failed: {str(e)}", exc_info=True)
    run_in_context()  # Execute a function with context
# Start the threadthread = (target=process_file_background)
()

Things to note

Only suitable for lightweight tasks, becauserequestThe object may be large, and copying will take up extra memory.
If the request has ended,requestMay fail, resulting in unpredictable behavior.

Summarize

plan	Applicable scenarios	advantage	shortcoming
Method 1 (pass in advance`user_id`）	Need to accurately record user operations	Clear code, avoid dependencies`request`	Function parameters need to be adjusted
Method 2 (optional`user_id`）	Logs can not be associated with users	High flexibility	Some logs may be missing user information
Method 3 (`copy_current_request_context`）	Simple task, complete`request`	Keep the complete request data	Possible memory usage

Best Practice Recommendations

Priority usage method 1 (pass in advanceuser_id) to avoid background thread dependenciesrequest。
If the log allows anonymous logging, use Method 2 to enhance robustness.
Use Method 3 only for simple tasks to avoid memory problems.

Expanding thinking

How to handle background tasks in conjunction with Celery?
- The Celery task has no Flask context by default and needs to be passed manually.user_idOr useflask-httpauthSuch as a plan.
Can it be usedgObjects store user information?
- gThe object is also in the request context, and the background thread cannot access it, so the data still needs to be extracted in advance.

The above is the detailed content of the analysis and solution of the request context problem in the Flask background thread. For more information about Flask background processing request context, please follow my other related articles!