跨文件获取数据库数据时出现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
相关产品推荐
相关产品推荐

