如何在Flask路由间持久化非JSON可序列化的BLE连接对象

首先看你代码里的几个核心问题：

• 缓存键名大小写不一致：ConnectUUT里用cache.set("BLEComm", BLEComm)存储对象，DisconnectUUT里却用cache.get("ble_comm")读取，大小写不匹配直接导致取不到实例。
• SimpleCache是进程内缓存，若服务器启用多进程模式（比如Waitress默认可能配置多进程），不同进程的缓存完全隔离，对象无法跨进程共享。
• 代码中未从请求参数中获取MAC_ID，直接调用会触发未定义变量错误。

针对非JSON可序列化对象的跨路由持久化，推荐以下几种可靠方案：

方案1：将实例绑定到Flask App全局对象 #

Flask的app对象是全局共享的，直接把BLE连接实例挂载到app上，彻底规避缓存的序列化和跨进程隔离问题。

修改后的代码：

from flask import Flask, request
from waitress import serve
from clsBLEcomm import clsBLEcomm

app = Flask(__name__)

# 初始化BLE连接实例到app全局属性
app.ble_comm = clsBLEcomm()

@app.route('/api/ConnectUUT', methods=['get'])
async def ConnectUUT():
    ErrorResponse = '0: OK'
    mac_id = request.args.get('MAC_ID')
    if not mac_id:
        return '1: Missing MAC_ID parameter', 400
    
    await app.ble_comm.ConnectUUT(MAC_ID=mac_id)
    return ErrorResponse

@app.route('/api/DisconnectUUT', methods=['get'])
async def DisconnectUUT():
    ErrorResponse = '0: OK'
    await app.ble_comm.DisconnectUUT()
    return ErrorResponse

if __name__ == '__main__':
    serve(app, port=5000)

方案2：用单例模式实现BLE通信类 #

让clsBLEcomm本身成为单例，无论在哪调用clsBLEcomm()都会返回同一个实例，从根源解决实例共享问题。

修改clsBLEcomm.py：

class clsBLEcomm:
    _instance = None

    def __new__(cls, *args, **kwargs):
        if not cls._instance:
            cls._instance = super().__new__(cls, *args, **kwargs)
        return cls._instance

    # 原有连接逻辑
    async def ConnectUUT(self, MAC_ID):
        # 你的BLE连接实现
        pass

    # 原有断开逻辑
    async def DisconnectUUT(self):
        # 你的BLE断开实现
        pass

对应Flask代码简化为：

from flask import Flask, request
from waitress import serve
from clsBLEcomm import clsBLEcomm

app = Flask(__name__)

@app.route('/api/ConnectUUT', methods=['get'])
async def ConnectUUT():
    ErrorResponse = '0: OK'
    mac_id = request.args.get('MAC_ID')
    if not mac_id:
        return '1: Missing MAC_ID parameter', 400
    
    ble_comm = clsBLEcomm()
    await ble_comm.ConnectUUT(MAC_ID=mac_id)
    return ErrorResponse

@app.route('/api/DisconnectUUT', methods=['get'])
async def DisconnectUUT():
    ErrorResponse = '0: OK'
    ble_comm = clsBLEcomm()
    await ble_comm.DisconnectUUT()
    return ErrorResponse

if __name__ == '__main__':
    serve(app, port=5000)

方案3：修正缓存用法（仅适合同进程服务器） #

如果坚持用Flask-Caching，先修正键名问题，同时强制服务器使用单进程模式：

修改后的缓存版代码：

from flask import Flask, request
from flask_caching import Cache
from waitress import serve
from clsBLEcomm import clsBLEcomm

config = {
    "DEBUG": True,
    "CACHE_TYPE": "SimpleCache",
    "CACHE_DEFAULT_TIMEOUT": 300 
}

app = Flask(__name__)
app.config.from_mapping(config)
cache = Cache(app)

@app.route('/api/ConnectUUT', methods=['get'])
async def ConnectUUT():
    ErrorResponse = '0: OK'
    mac_id = request.args.get('MAC_ID')
    if not mac_id:
        return '1: Missing MAC_ID parameter', 400
    
    # 统一使用小写键名
    BLEComm = cache.get("ble_comm_instance")
    if BLEComm is None:
        BLEComm = clsBLEcomm()
        cache.set("ble_comm_instance", BLEComm)
    
    await BLEComm.ConnectUUT(MAC_ID=mac_id)
    return ErrorResponse

@app.route('/api/DisconnectUUT', methods=['get'])
async def DisconnectUUT():
    ErrorResponse = '0: OK'
    # 用相同键名读取实例
    BLEComm = cache.get("ble_comm_instance")
    if BLEComm is None:
        return '1: No active BLE connection', 400
    
    await BLEComm.DisconnectUUT()
    # 断开后清除缓存中的实例
    cache.delete("ble_comm_instance")
    return ErrorResponse

if __name__ == '__main__':
    # 指定Waitress用单进程，避免多进程缓存隔离
    serve(app, port=5000, threads=4, processes=1)

注意：若服务器启用多进程模式，SimpleCache无法跨进程共享对象，此时方案1或方案2是更可靠的选择。

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