在Flask中获取所有客户端活跃Session对象列表的方法

首先明确核心前提：Flask默认的客户端Session（签名Cookie模式）下，服务端不会存储任何Session数据——所有会话内容都加密后存在用户浏览器的Cookie里，这种场景下服务端根本无法直接获取所有活跃Session列表。只有当你使用服务器端Session存储（通过Flask-Session配置Redis、Memcached、文件系统、数据库等后端）时，才能从存储介质中读取所有活跃Session数据。

以下是不同存储后端的具体实现方式：

1. Redis作为Session存储后端（Flask-Session配置） #

若Session存在Redis中，直接通过Redis的键遍历功能获取所有Session键，再读取并反序列化对应值：

from flask import Flask
from flask_session import Session
import redis
from flask_session.serializers import pickle

app = Flask(__name__)
app.config['SESSION_TYPE'] = 'redis'
app.config['SESSION_REDIS'] = redis.Redis(host='localhost', port=6379)
Session(app)

def get_all_active_sessions():
    redis_conn = app.config['SESSION_REDIS']
    # Flask-Session默认Session键前缀为'session:'
    session_keys = redis_conn.keys('session:*')
    active_sessions = []
    for key in session_keys:
        session_data = redis_conn.get(key)
        try:
            # 反序列化得到会话字典（与请求上下文的session对象内容一致）
            session_obj = pickle.loads(session_data)
            active_sessions.append(session_obj)
        except:
            # 跳过过期或损坏的Session数据
            continue
    return active_sessions

后台线程直接调用get_all_active_sessions()即可拿到所有活跃会话数据列表。

2. 文件系统作为Session存储后端 #

若配置SESSION_TYPE='filesystem'，Flask-Session会将每个Session存入单独文件（默认目录为flask_session），遍历目录读取文件并反序列化：

import os
from flask_session.serializers import pickle

def get_all_active_sessions():
    session_dir = app.config.get('SESSION_FILE_DIR', 'flask_session')
    active_sessions = []
    for filename in os.listdir(session_dir):
        file_path = os.path.join(session_dir, filename)
        if os.path.isfile(file_path):
            with open(file_path, 'rb') as f:
                session_data = f.read()
            try:
                session_obj = pickle.loads(session_data)
                active_sessions.append(session_obj)
            except:
                continue
    return active_sessions

3. 数据库作为Session存储后端（如SQLAlchemy） #

若使用数据库存储（Flask-Session的sqlalchemy类型），直接查询sessions表中未过期的记录：

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from datetime import datetime
from flask_session.serializers import pickle

# 与Flask-Session配置一致的数据库连接
engine = create_engine(app.config['SESSION_SQLALCHEMY'])
DbSession = sessionmaker(bind=engine)
db_session = DbSession()

def get_all_active_sessions():
    # 过滤已过期的会话记录
    session_records = db_session.execute(
        "SELECT data FROM sessions WHERE expire_time > :now", 
        {"now": datetime.utcnow()}
    ).fetchall()
    active_sessions = []
    for record in session_records:
        try:
            session_obj = pickle.loads(record[0])
            active_sessions.append(session_obj)
        except:
            continue
    return active_sessions

关键注意事项 #

• 你获取到的是Session存储的字典数据，而非请求上下文里的session对象，但两者包含的会话内容完全一致。
• 必须处理过期或损坏的Session数据，避免读取无效内容导致报错。
• 若使用默认客户端Cookie Session，无法直接获取所有Session，只能通过间接方式追踪活跃用户（比如登录时记录用户ID到数据库，定期清理长期无活动的用户）。

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