Flask-SQLAlchemy多库下db.session.execute查询Table报错问题咨询

我尝试在Flask-SQLAlchemy中使用多数据库，已为Table指定bind_key，使用Model和新创建的session查询均正常，但db.session.execute无法识别该Table元数据中的bind_key，抛出以下错误：

$ python app.py 
model [] <-- works
new session [] <-- works
Traceback (most recent call last):
  File "C:\Users\user\projects\flask-sqlalchemy\app.py", line 32, in <module>
    print('main session', db.session.execute(select(user)).all()) # <- UnboundExecutionError
  File "C:\Users\user\projects\flask-sqlalchemy\venv\lib\site-packages\sqlalchemy\orm\scoping.py", line 778, in execute
    return self._proxied.execute(
  File "C:\Users\user\projects\flask-sqlalchemy\venv\lib\site-packages\sqlalchemy\orm\session.py", line 2306, in execute
    return self._execute_internal(
  File "C:\Users\user\projects\flask-sqlalchemy\venv\lib\site-packages\sqlalchemy\orm\session.py", line 2179, in _execute_internal
    bind = self.get_bind(**bind_arguments)
  File "C:\Users\user\projects\flask-sqlalchemy\venv\lib\site-packages\flask_sqlalchemy\session.py", line 78, in get_bind
    return super().get_bind(mapper=mapper, clause=clause, bind=bind, **kwargs)
  File "C:\Users\user\projects\flask-sqlalchemy\venv\lib\site-packages\sqlalchemy\orm\session.py", line 2786, in get_bind
    raise sa_exc.UnboundExecutionError(
sqlalchemy.exc.UnboundExecutionError: Could not locate a bind configured on SQL expression or this Session.

代码示例 #

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy.orm import scoped_session, sessionmaker
from sqlalchemy import Column, Integer, String, select

app = Flask(__name__)
# app.config["SQLALCHEMY_DATABASE_URI"] = "sqlite:///main.sqlite"
app.config["SQLALCHEMY_BINDS"] = {"secondary": "sqlite:///secondary.sqlite"}

db = SQLAlchemy(app)

class User(db.Model):
    __bind_key__ = "secondary"
    user_id = Column(Integer, primary_key=True)
    user_name = Column(String)

user = db.Table(
    "user_table",
    Column("user_id", Integer, primary_key=True),
    Column("user_name", String),
    bind_key="secondary",
)

with app.app_context():
    db.create_all("secondary")

    print('model', db.session.execute(select(User)).all()) # <- works

    new_session = scoped_session(sessionmaker(bind=db.engines['secondary']))

    print('new session', new_session.execute(select(user)).all()) # <- works
    print('main session', db.session.execute(select(user)).all()) # <- UnboundExecutionError

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

测试步骤 #

执行pip install -U Flask-SQLAlchemy，然后运行python app.py即可复现问题。

疑问 #

这是我的操作有误，还是存在Bug？我认为无需手动为每个数据库创建session。

参考文档说明：

带有bind_key的模型和表会注册到对应的元数据中，会话会使用对应的引擎查询它们。

内容的提问来源于stack exchange，提问作者Andrew O.