You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

跨文件获取数据库数据时出现KeyError异常的问题求助

问题描述

尝试获取两个数据库的条目数量时,第二个数据库计数功能正常,但第一个数据库抛出以下错误:

KeyError: <weakref at 0x000001E85C863330; to "Flask" at 0x000001E8397750D0>

程序已做简化处理,移除的Get、Post、Delete等功能均正常运行。项目包含三个文件:

server1.py

app = Flask(__name__)

api = Api(app)

app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///emp.db'
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db = SQLAlchemy(app)

class Value(db.Model):
    id = db.Column(db.Integer, primary_key=True)
    value = db.Column(db.Integer, nullable=False)

class GetCount(Resource):
    @staticmethod
    def count():
        count = Value.query.count()
        return count

server2.py

app2 = Flask(__name__)

api2 = Api(app2)

app2.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///emp2.db'
app2.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False

db2 = SQLAlchemy(app2)


class Value2(db2.Model):
    id = db2.Column(db2.Integer, primary_key=True)
    value = db2.Column(db2.Integer, nullable=False)

class GetCount2(Resource):
    @staticmethod
    def count():
        count = Value2.query.count()
        return count

masternode.py

import time
from server1 import app, Value
from server2 import app2, Value2

app.app_context().push()
app2.app_context().push()

while True:
    c = Value.query.count()
    c2 = Value2.query.count()
    print(c, c2)
    time.sleep(1)

启动程序时触发上述错误,但删除masternode.py中的c = Value.query.count()语句后,程序可正常持续输出1 1 1 1...。

完整错误栈:

Traceback (most recent call last):
  File "C:\Users\Sergio\Desktop\Домашка\FlaskTest\masternode.py", line 15, in <module>
    c1 = Value.query.count()
         ^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Sergio\AppData\Local\Programs\Python\Python311\Lib\site-packages\sqlalchemy\orm\query.py", line 3175, in count
    return self._from_self(col).enable_eagerloads(False).scalar()
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Sergio\AppData\Local\Programs\Python\Python311\Lib\site-packages\sqlalchemy\orm\query.py", line 2892, in scalar
    ret = self.one()
          ^^^^^^^^^^
  File "C:\Users\Sergio\AppData\Local\Programs\Python\Python311\Lib\site-packages\sqlalchemy\orm\query.py", line 2869, in one
    return self._iter().one()
           ^^^^^^^^^^^^
  File "C:\Users\Sergio\AppData\Local\Programs\Python\Python311\Lib\site-packages\sqlalchemy\orm\query.py", line 2915, in _iter
    result = self.session.execute(
             ^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Sergio\AppData\Local\Programs\Python\Python311\Lib\site-packages\sqlalchemy\orm\session.py", line 1702, in execute
    bind = self.get_bind(**bind_arguments)
           ^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
  File "C:\Users\Sergio\AppData\Local\Programs\Python\Python311\Lib\site-packages\flask_sqlalchemy\session.py", line 61, in get_bind
    engines = self._db.engines
              ^^^^^^^^^^^^^^^^
  File "C:\Users\Sergio\AppData\Local\Programs\Python\Python311\Lib\site-packages\flask_sqlalchemy\extension.py", line 629, in engines
    return self._app_engines[app]
           ~~~~~~~~~~~~~~~~~^^^^^
  File "C:\Users\Sergio\AppData\Local\Programs\Python\Python311\Lib\weakref.py", line 415, in __getitem__
    return self.data[ref(key)]
           ~~~~~~~~~^^^^^^^^^^
KeyError: <weakref at 0x000001E85C863330; to 'Flask' at 0x000001E8397750D0>
原因分析与解决方法

错误原因

问题出在Flask-SQLAlchemy的上下文管理机制:连续调用push()推入两个应用上下文时,后推入的app2上下文会覆盖前一个app上下文,导致第一个应用的上下文未被正确保留,SQLAlchemy无法找到对应的Flask应用实例,从而抛出弱引用的KeyError。

解决方法1:使用上下文管理器隔离操作(推荐)

不要一次性推入两个上下文,而是在每次查询时,用with语句确保当前操作在对应的应用上下文中执行:

修改后的masternode.py:

import time
from server1 import app, Value
from server2 import app2, Value2

while True:
    # 在server1的上下文环境中执行查询
    with app.app_context():
        c = Value.query.count()
    # 在server2的上下文环境中执行查询
    with app2.app_context():
        c2 = Value2.query.count()
    print(c, c2)
    time.sleep(1)

解决方法2:手动管理上下文栈

如果需要同时保留两个上下文,可以在每次查询前手动切换到对应上下文:

import time
from server1 import app, Value
from server2 import app2, Value2

while True:
    # 切换到app的上下文
    ctx1 = app.app_context()
    ctx1.push()
    c = Value.query.count()
    ctx1.pop()
    
    # 切换到app2的上下文
    ctx2 = app2.app_context()
    ctx2.push()
    c2 = Value2.query.count()
    ctx2.pop()
    
    print(c, c2)
    time.sleep(1)

为什么第二个数据库能正常工作?

因为最后推入的是app2的上下文,执行Value2.query.count()时,当前活跃的上下文是app2的,能正确匹配到对应的SQLAlchemy引擎;而第一个数据库的查询尝试使用已被覆盖的app上下文,导致找不到对应的弱引用实例。

内容的提问来源于stack exchange,提问作者Sergio

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.08.11 05:01:11