如何在Flask中遍历所有session并批量登出非管理员用户？

问题解答 #

flask_session扩展提供的Session类本身没有内置存储所有会话的公共列表/属性可供直接调用。
该扩展本质是会话适配层，所有会话数据实际存储在你配置的SESSION_TYPE对应的后端服务中，你需要直接操作对应后端存储遍历、筛选、删除会话，以下是不同常见存储场景的实现方案：

场景1：SESSION_TYPE = 'redis' #

from flask import current_app

# 获取flask_session绑定的Redis客户端实例
redis_client = current_app.extensions['session'].redis
# flask_session默认会话键前缀为'session:'，如果自定义了SESSION_KEY_PREFIX请替换
session_keys = redis_client.keys('session:*')
serializer = current_app.extensions['session'].serializer

for key in session_keys:
    session_data = serializer.loads(redis_client.get(key))
    # 跳过管理员会话
    if session_data.get('is_admin') == True:
        continue
    # 删除非管理员会话实现登出
    redis_client.delete(key)

注意：生产环境会话量较大时建议用scan迭代替代keys命令，避免阻塞Redis实例。

场景2：SESSION_TYPE = 'filesystem' #

import os
from flask import current_app

session_dir = current_app.config.get('SESSION_FILE_DIR', 'flask_session')
serializer = current_app.extensions['session'].serializer

for filename in os.listdir(session_dir):
    file_path = os.path.join(session_dir, filename)
    with open(file_path, 'rb') as f:
        session_data = serializer.load(f)
    if session_data.get('is_admin') == True:
        continue
    os.remove(file_path)

场景3：SESSION_TYPE = 'sqlalchemy' #

from flask import current_app

session_store = current_app.extensions['session'].store
# 查询所有存储的会话
all_sessions = session_store.session_class.query.all()

for s in all_sessions:
    session_data = session_store.serializer.loads(s.data)
    if session_data.get('is_admin') == True:
        continue
    session_store.db.session.delete(s)
# 提交删除操作
session_store.db.session.commit()

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