每个会话有多个数据库后端.使用sqlhemy时执行

Question

我有这个数据库连接在一个快速的API应用程序。我对同一个会话使用多个绑定，我遵循这里的文档:查询db:

method1:工作

db.query(Model).all()

method2:不工作，抛出以下错误:

db.execute("SELECT * from ...");
Exception has occurred: UnboundExecutionError
Could not locate a bind configured on SQL expression or this Session.

这是数据库连接代码…你能帮我启动method2吗?

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.automap import automap_base
from sqlalchemy import MetaData
from app.settings import (
db1_DATABASE_URL,
db2_DATABASE_URL
)

engine = create_engine(
db1_DATABASE_URL,
echo=False,
pool_recycle=1800,
pool_timeout=20,
pool_pre_ping=True,
pool_size=50,
max_overflow=100,
)
engine2 = create_engine(db2_DATABASE_URL)

Base = declarative_base()
mroi_Base = automap_base()
mroi_Base.prepare(engine2, reflect=True)
SessionLocal = sessionmaker(autocommit=False, autoflush=False)
SessionLocal.configure(binds={Base: engine, mroi_Base: engine2})

def get_db():
db = None
try:
db = SessionLocal()
yield db
finally:
db.close()

def get_db2():
db2 = None
try:
db2 = SessionLocal()
yield db2
finally:
db2.close()

Answer 1

删除下面提到的行，因为我们将在创建各自的会话实例的同时配置引擎。

SessionLocal.configure(binds={Base: engine, mroi_Base: engine2})

用各自的引擎值初始化会话类。

def get_db():
db = None
try:
db = SessionLocal(engine=engine)
yield db
finally:
db.close()

def get_db2():
db2 = None
try:
db2 = SessionLocal(engine=engine2)
yield db2
finally:
db2.close()