解决Flask热重载中“操作尝试在非套接字对象上进行”的OSError

import logging
import threading
from flask_cors import CORS
from flask import Flask, request

# 假设 PostgreDB 是一个自定义的数据库连接类，
# 并在其内部启动了一个独立的线程来处理请求队列。
class PostgreDB:
    def __init__(self):
        logging.info("Initializing PostgreDB instance...")
        self.connection = self._connect_to_db()
        self.request_queue = []
        self.worker_thread = threading.Thread(target=self._process_requests, daemon=True)
        self.worker_thread.start()

    def _connect_to_db(self):
        # 模拟数据库连接
        return "DB_CONNECTION_OBJECT"

    def _process_requests(self):
        while True:
            if self.request_queue:
                # 处理队列中的请求
                pass
            # 模拟工作
            threading.Event().wait(0.1)

    def close(self):
        logging.info("Closing PostgreDB instance...")
        # 清理数据库连接和线程
        pass

app = Flask(__name__)

# 全局初始化数据库实例
# 这是导致问题的关键点：热重载会创建多个此实例
db = PostgreDB()

@app.route('/')
def index():
    # 使用db实例
    return "Hurray!"

if __name__ == '__main__':
    app.run(host='0.0.0.0', port=5500, debug=True)

import logging
import threading
from flask_cors import CORS
from flask import Flask, request, g # 引入 g 对象

# 假设 MyDB 是一个自定义的数据库连接类，
# 确保其 __del__ 或 close 方法能正确清理资源，包括可能存在的线程。
class MyDB:
    def __init__(self):
        logging.info("Initializing MyDB instance...")
        self.connection = self._connect_to_db()
        # 如果有独立线程，确保线程的生命周期与 MyDB 实例绑定
        # 并在 close() 或 __del__() 中正确终止
        # self.worker_thread = threading.Thread(...)
        # self.worker_thread.start()

    def _connect_to_db(self):
        # 模拟数据库连接
        return "DB_CONNECTION_OBJECT"

    def get_name(self, user_info):
        # 模拟从数据库获取数据
        return f"User_{user_info['id']}"

    def close(self):
        logging.info("Closing MyDB instance...")
        # 确保在这里关闭数据库连接并终止任何相关线程
        if self.connection:
            # self.connection.close() # 实际关闭连接
            self.connection = None
        # if self.worker_thread and self.worker_thread.is_alive():
        #     self.worker_thread.join(timeout=1) # 等待线程结束

# 数据库获取函数
def get_db():
    """
    此函数将数据库实例插入到Flask的全局变量命名空间 `g` 中，
    该实例在应用程序上下文销毁后关闭。
    """
    if 'db' not in g:
        g.db = MyDB()  # 创建一个新的数据库连接
    return g.db

# 创建Flask应用工厂函数
def create_app():
    app = Flask(__name__)
    CORS(app) # 示例：添加CORS支持

    # 在每个请求之前执行：确保 g.db 被设置
    @app.before_request
    def before_request():
        g.db = get_db()

    # 在应用程序上下文销毁时执行：清理数据库连接
    @app.teardown_appcontext
    def teardown_db(exception):
        db_instance: MyDB | None = g.pop('db', None)
        if db_instance is not None:
            db_instance.close()

    # 注册FLASK路由
    @app.route('/')
    def index():
        name = g.db.get_name({"id": 123}) # 如何在代码中使用 g.db
        return f"Hello, {name}!"

    return app

if __name__ == '__main__':
    app = create_app()
    app.run(host='0.0.0.0', port=5500, debug=True)