请求上下文外操作报错：如何在普通函数中删除Flask Session

我需要实现一个功能：当页面关闭且用户长时间未重新打开时，删除其用户Session。测试方式为打开网页后关闭，保持服务器运行。目前断开连接与重新登录功能正常，但执行Session清除操作时触发RuntimeError: Working outside of request context错误。尝试过传递Session对象、使用app.app_context()，均未解决问题，怀疑是清除函数没有Flask/Socket装饰器导致无法处理请求。

控制台报错日志 #

init
poll
poll
poll
poll
poll
going to clear
File "/Volumes/src/testing/test.py", line 46, in sessiondelete
clear_session(session)
File "/Volumes/src/testing/test.py", line 35, in clear_session
session.clear()
RuntimeError: Working outside of request context.

This typically means that you attempted to use functionality that needed
an active HTTP request. Consult the documentation on testing for
information about how to avoid this problem.

复现代码 #

Flask后端代码

from flask import Flask, session, render_template, url_for, redirect
from flask_socketio import SocketIO, join_room
from flask_session import Session
from threading import Thread
from time import sleep
app = Flask(__name__)
app.config['SECRET_KEY'] = 'temporary2'
app.config['SESSION_TYPE'] = 'filesystem'
Session(app)
socketio = SocketIO(app, manage_session=False)
deletelist = {}
@app.route('/')
def index():
    return render_template('wait.html')
@app.route('/dir')
def foo():
    if "name" in session:
        name = session['name']
        return f"Hello, {name}!"
    else:
        return redirect(url_for("index"))

@socketio.on('log')
def connection(request):
    if "name" not in session:
        join_room('waiting')
        session['name'] = 'craig'
        session.modified = True
        socketio.emit('redirect', url_for('foo'), room='waiting')
    else:
        deletelist[session['name']]= False # 用户短时间内重新登录时取消删除
        print('relogin')
def clear_session():
    with app.app_context():
        session.clear()
        print('Session cleared')
        
@socketio.on('disconnect')
def disconnect():
    def sessiondelete(sess,t):
        if deletelist[sess]==False:
            print('cancelled')
            del deletelist[sess]
        elif t == 5:# 测试用5秒，实际可设置更长时间
            print('cleared')
            clear_session()
            del deletelist[sess]
        elif deletelist[sess]==True:
            print('poll')
            sleep(1)
            sessiondelete(sess,t+1)
    deletelist[session['name']]= True # Session对象不可哈希，所以存用户名
    print('init')
    worker = Thread(target=sessiondelete, args=(session['name'],0))# 开线程不阻塞其他用户
    worker.start()
if __name__ == '__main__':   
   socketio.run(app, port=5200, debug=True)

前端HTML代码（wait.html）

<!DOCTYPE html>
<html lang="en">
    <head>
        <meta charset="UTF-8">
    </head>
    <body>
        Waiting...
        <script src="https://cdnjs.cloudflare.com/ajax/libs/socket.io/4.0.1/socket.io.js"></script>
        <script>
        const socket = io();
        socket.emit("log", { data: "connection" });
        socket.on('redirect', (dest) => {
            window.location = dest;
        });
        </script>
    </body>
</html>

解决方案 #

问题核心在于：Flask的session对象是请求上下文绑定的，脱离请求上下文直接调用session.clear()必然报错。正确做法是通过用户的Session ID，直接操作flask-session的底层存储来删除Session，无需依赖请求上下文。

修改后的后端代码

from flask import Flask, session, render_template, url_for, redirect
from flask_socketio import SocketIO, join_room
from flask_session import Session
from threading import Thread
from time import sleep
app = Flask(__name__)
app.config['SECRET_KEY'] = 'temporary2'
app.config['SESSION_TYPE'] = 'filesystem'
Session(app)
socketio = SocketIO(app, manage_session=False)
# 改为存储Session ID和删除状态，避免用户名重复问题
deletelist = {}

@app.route('/')
def index():
    return render_template('wait.html')

@app.route('/dir')
def foo():
    if "name" in session:
        name = session['name']
        return f"Hello, {name}!"
    else:
        return redirect(url_for("index"))

@socketio.on('log')
def connection(request):
    if "name" not in session:
        join_room('waiting')
        session['name'] = 'craig'
        session.modified = True
        socketio.emit('redirect', url_for('foo'), room='waiting')
    else:
        # 用户重新登录，取消对应Session的删除计划
        session_id = session.sid
        if session_id in deletelist:
            deletelist[session_id] = False
        print('relogin')

def clear_session(session_id):
    # 获取session存储接口，直接操作底层存储
    session_interface = app.session_interface
    session_interface.delete_session(app, session_id)
    print(f'Session {session_id} cleared')
        
@socketio.on('disconnect')
def disconnect():
    def sessiondelete(session_id, t):
        if deletelist.get(session_id) == False:
            print('cancelled')
            del deletelist[session_id]
        elif t == 5: # 测试用5秒，实际可设置更长时间
            print('cleared')
            clear_session(session_id)
            del deletelist[session_id]
        elif deletelist.get(session_id) == True:
            print('poll')
            sleep(1)
            sessiondelete(session_id, t+1)
    
    session_id = session.sid
    deletelist[session_id] = True
    print('init')
    worker = Thread(target=sessiondelete, args=(session_id, 0))
    worker.start()

if __name__ == '__main__':   
   socketio.run(app, port=5200, debug=True)

修改说明

1. 用session.sid（唯一Session ID）替代用户名作为deletelist的键，避免用户名重复导致的错误
2. clear_session函数不再依赖请求上下文，通过app.session_interface直接调用delete_session方法删除指定ID的Session
3. 用户重新登录时，通过Session ID找到对应的删除计划并取消

内容的提问来源于stack exchange，提问作者user12967009