How can I access current_user outside of an route. FLASK API

Question

Im trying to make a polling mechanism, so im making a background process using "ThreadPoolExecutor", I added a status boolean to my users table so they will only be able to send 1 request at time, but i´ve ran into a problem where I cant change the current_user.status to False inside of the background process is over since its outside of an route.

def background_translation(file_content, w_model, transcription, model, pitch, speech_rate, user_id):
    try:
        srt_file = whisper_transcribe(file_content, w_model, transcription)
        audio = text_to_speech(srt_file, model, pitch, speech_rate)
        output = add_stream(audio, file_content)

        # Save as user_id.mp4
        destination = os.path.join(CONTENT_FOLDER, f"{user_id}.mp4")
        shutil.move(output, destination)

        print(f"Translation complete: saved to {destination}")

    except Exception as e:
        print("BGT error during translation:", e)


u/bp.route('/translator', methods=['POST'])
u/login_or_token_required
def translator(user):

    #inputs...

    user_id = current_user.id

    start_process(current_user)

    file_extension = secure_filename(filepath.filename)
    file_content = os.path.join(UPLOAD_FOLDER, file_extension)
    filepath.save(file_content)


    print("executor.submit")
    executor.submit(
        background_translation,
        file_content,
        w_model,
        transcription,
        model,
        pitch,
        speech_rate,
        user_id
    )
    print("Sent")

    return jsonify({
        "message": "Translation started",
    }), 200

I tried to access the query by user_id

user = User.query.get(user_id)

Answer 1

When you call executor.submit you submit the job to another thread without context of the Flask application (same as if you create a script to call directly background_translation ). This is usually done behind the scene by Flask when responding to an HTTP query.

One way to do that is using app.app_context

For example modify background_translation :

def background_translation(file_content, w_model, transcription, model, pitch, speech_rate, user_id):
    
    # creating the app (using application factories)
    str_config = "THIS_CONFIG_NAME"
    app = create_app(str_config) #create the Flask app
    
    # Now inject the context (as you are not responding to an HTTP request)
    with app.app_context():
    ## a more advanced version (for example with translation with Babel):
    ## with app.request_context({'wsgi.url_scheme': 'https', 'SERVER_PORT': 443, 'SERVER_NAME': 'myServerNameToReplace.test.com', 'REQUEST_METHOD': "GET",'wsgi.errors':stream_logger_if_needed}):

        ## Just in cases to add a database session :
        # with get_db_session() as db_session: # look into that if you want to also use SQLAlchemy for database.

        try:
            srt_file = whisper_transcribe(file_content, w_model, transcription)
            audio = text_to_speech(srt_file, model, pitch, speech_rate)
            output = add_stream(audio, file_content)
    
            # Save as user_id.mp4
            destination = os.path.join(CONTENT_FOLDER, f"{user_id}.mp4")
            shutil.move(output, destination)
    
            print(f"Translation complete: saved to {destination}")
    
        except Exception as e:
            print("BGT error during translation:", e)

I advice you to move background_translation to another module to make this more clear.