Flask-Login中为current_user设置非数据库持久化属性的问题

问题场景 #

我在开发基于Flask-Login的应用时，尝试给flask_login.current_user添加一个无需存入数据库的跨请求持久化属性——一个复杂的Controller类实例。该实例包含多个子对象和数据库连接，无法序列化，且初始化耗时极长，重复初始化会严重拖慢性能。

我通过以下代码给current_user赋值：

@leagues.route('/admin-attend', methods=["POST"])
@login_required
def admin_attending():
    """
    When the admin opens up its league.
    """
    json_file = request.get_json()
    current_user.controller = Controller(json_file["league_id"])
    json_return["status"] = "success"
    return redirect('/session/')

并在另一个路由中访问该实例：

@session_page.route('/', methods=['GET'])
@admin_required
def session():
    """
    Renders the session page, and loads in all of the data.
    """

    com = current_user.controller
    present_players, rounds, absent_players = getval(com)
    return render_template("session.html", players=present_players, 
                           absent=absent_players, 
                           rounds=rounds)

本地Windows开发环境（Python 3.10.9，app.run(debug=True)）中功能完全正常，但部署到Ubuntu服务器（Python 3.9.12，WSGI运行）后，出现属性错误，提示current_user不存在controller属性。

我的User类、数据库映射及user_loader代码如下：

User类 #

class User(UserMixin):
    """ 
    This class is mapped to the user table in the database.
    As a client navigates the browser it tracks the admin priveleges.
    The users reciever and controller properties allow a client to remain 
    using the same reciever/controller while navigating without 
    reinitializing.
    """
    def __init__(self, email, password, session):
        self.email = email
        self.password = password
        self.session = session
        self.is_admin = False
        self.current_league = None
        self.reciever = None
        self.controller = None
        
    def get_id(self):
        """ 
        Gets the ID of the player.
        """
        return self.player_id

数据库映射 #

# Map User Class to User Table.
metadata = MetaData()
user_table = Table('user', metadata, autoload=True,autoload_with=engine)
mapper(User, user_table)

user_loader #

# Loads the current_user.
@login_manager.user_loader
def load_user(id):
    return session.query(User).get(id)

注：Controller类无法存入Cookie或JSON序列化，不能使用Flask的session存储。

问题原因分析 #

1. 本地与生产环境的运行模式差异
   • 本地debug模式下，Flask默认以单线程运行，所有请求复用同一个进程和内存空间。给current_user添加的controller属性会保存在内存中的User实例里，下一个请求加载的current_user仍是同一个内存对象，因此能正常访问属性。
   • 生产环境的WSGI服务器（如Gunicorn、uWSGI）通常采用多进程/多线程模式，不同请求可能由不同进程处理。每次请求时，user_loader都会从数据库重新加载全新的User实例，之前进程中给User实例添加的controller属性无法跨进程共享，导致属性不存在的错误。
2. 对象生命周期限制
   即使是单进程WSGI模式，服务器也可能在请求结束后回收或重置对象，临时添加的属性会随对象销毁而丢失。

可行解决方案 #

方案1：服务器端会话存储（基于用户ID关联） #

用全局内存字典（单进程场景）或分布式存储（如Redis，多进程/分布式场景）存储Controller实例，以用户ID为关联键：

# 单进程场景用全局字典
USER_CONTROLLERS = {}

# 多进程/分布式场景用Redis（需先安装redis库）
# import redis
# redis_client = redis.Redis(host='localhost', port=6379, db=0)

@leagues.route('/admin-attend', methods=["POST"])
@login_required
def admin_attending():
    json_file = request.get_json()
    league_id = json_file["league_id"]
    controller = Controller(league_id)
    
    # 单进程存储方式
    USER_CONTROLLERS[current_user.get_id()] = controller
    
    # 多进程/分布式存储方式（若Controller可pickle序列化）
    # redis_client.set(f"user_controller:{current_user.get_id()}", pickle.dumps(controller))
    # 更安全的方式：仅存储初始化参数，避免序列化风险
    # redis_client.set(f"user_league_id:{current_user.get_id()}", league_id)
    
    return redirect('/session/')

@session_page.route('/', methods=['GET'])
@admin_required
def session():
    user_id = current_user.get_id()
    
    # 单进程读取方式
    controller = USER_CONTROLLERS.get(user_id)
    if not controller:
        return redirect('/leagues/')
    
    # 多进程/分布式读取方式（读取league_id后初始化）
    # league_id = redis_client.get(f"user_league_id:{user_id}")
    # if not league_id:
    #     return redirect('/leagues/')
    # controller = Controller(league_id.decode('utf-8'))
    
    present_players, rounds, absent_players = getval(controller)
    return render_template("session.html", players=present_players, 
                           absent=absent_players, 
                           rounds=rounds)

注意：使用pickle序列化存入Redis存在安全风险，仅建议在可信环境中使用；优先选择存储初始化参数（如league_id），再按需初始化。

方案2：进程内缓存（LRU Cache） #

用functools.lru_cache缓存Controller的初始化方法，以league_id为缓存键，减少重复初始化次数：

from functools import lru_cache

# 设置缓存上限，避免内存溢出
@lru_cache(maxsize=32)
def get_controller(league_id):
    return Controller(league_id)

@session_page.route('/', methods=['GET'])
@admin_required
def session():
    # 假设league_id已存入User的current_league字段
    league_id = current_user.current_league
    if not league_id:
        return redirect('/leagues/')
    
    # 复用缓存的Controller实例
    controller = get_controller(league_id)
    present_players, rounds, absent_players = getval(controller)
    return render_template("session.html", players=present_players, 
                           absent=absent_players, 
                           rounds=rounds)

注意：需确保Controller实例是线程安全的，避免多线程请求同时访问时出现冲突。

方案3：自定义User属性加载逻辑 #

修改user_loader，在加载User实例时从全局存储中恢复controller属性：

USER_CONTROLLERS = {}

@login_manager.user_loader
def load_user(id):
    user = session.query(User).get(id)
    # 从全局存储中恢复controller属性
    user.controller = USER_CONTROLLERS.get(id)
    return user

@leagues.route('/admin-attend', methods=["POST"])
@login_required
def admin_attending():
    json_file = request.get_json()
    controller = Controller(json_file["league_id"])
    current_user.controller = controller
    # 同步到全局存储
    USER_CONTROLLERS[current_user.get_id()] = controller
    return redirect('/session/')

该方案仅适用于单进程WSGI环境，多进程场景需替换为Redis等分布式存储。

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