使用 sqlalchemy 时每个会话有多个数据库后端以及 db.execute

Question

我在一个快速 API 应用程序中有这个数据库连接。我在同一个会话中使用多个绑定，我遵循了文档 here：查询数据库：

方法一：有效

db.query(Model).all()

方法 2：无效并引发以下错误：

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})

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

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()

使用 sqlalchemy 时每个会话有多个数据库后端以及 db.execute

multiple database backends per session along with db.execute when using sqlaclhemy

python

sqlalchemy