SQLAlchemy测试中如何正确绑定双数据库连接以支持跨库操作

核心问题修正：把DSN字符串替换为Engine实例 #

你碰到的'str' object has no attribute 'connect'错误，本质是engines()返回的是字符串，而SQLAlchemy的Session必须绑定Engine实例才能正常工作。先修改database.py的engines()函数，返回真正的Engine对象：

# database.py
from sqlalchemy import create_engine

def engines():
    # 替换为你的实际数据库连接串
    engine_db1 = create_engine("mysql+pymysql://user:pass@localhost/db1")
    engine_db2 = create_engine("postgresql://user:pass@localhost/db2")
    return {"db1": engine_db1, "db2": engine_db2}

# 提前初始化全局引擎实例，避免重复创建
GLOBAL_ENGINES = engines()

双库Session绑定的两种实用方案 #

方案1：给模型固定绑定引擎（适合分库明确的场景）

如果你的模型本来就分属两个数据库，直接在模型类中指定__bind_key__，Session会自动路由到对应引擎：

# models.py 示例
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy import Column, Integer, String

Base = declarative_base()

class User(Base):
    __tablename__ = "users"
    __bind_key__ = "db1"  # 指定绑定到db1引擎
    id = Column(Integer, primary_key=True)
    username = Column(String(50))

class Order(Base):
    __tablename__ = "orders"
    __bind_key__ = "db2"  # 指定绑定到db2引擎
    id = Column(Integer, primary_key=True)
    user_id = Column(Integer)

然后在conftest.py中创建支持多绑定的Session fixture：

# conftest.py
import pytest
from sqlalchemy.orm import sessionmaker
from database import GLOBAL_ENGINES
from models import Base

@pytest.fixture(scope="session")
def init_databases():
    # 测试环境初始化表（生产环境请勿随意执行）
    for engine in GLOBAL_ENGINES.values():
        Base.metadata.create_all(bind=engine)
    yield
    # 测试结束后清理表
    for engine in GLOBAL_ENGINES.values():
        Base.metadata.drop_all(bind=engine)

@pytest.fixture(scope="function")
def db_session(init_databases):
    # 创建支持多绑定的Session工厂
    Session = sessionmaker(binds=GLOBAL_ENGINES)
    session = Session()
    try:
        yield session
    finally:
        session.rollback()
        session.close()

方案2：手动切换引擎（适合动态跨库查询场景）

如果需要在同一个Session中灵活操作两个库，可以手动切换绑定的引擎，或者直接用引擎执行原生SQL：

# 测试用例示例
def test_cross_database_query(db_session):
    # 从db1查询用户数据
    target_user = db_session.query(User).filter_by(id=1).first()
    
    # 手动切换到db2引擎查询关联订单
    db_session.bind = GLOBAL_ENGINES["db2"]
    user_orders = db_session.query(Order).filter_by(user_id=target_user.id).all()
    
    # 也可以直接用引擎执行原生SQL，更灵活
    raw_result = GLOBAL_ENGINES["db2"].execute(
        "SELECT COUNT(*) FROM orders WHERE user_id = :uid", 
        {"uid": target_user.id}
    )
    order_count = raw_result.scalar()
    
    assert len(user_orders) == order_count

验证运行 #

修改完成后重新执行测试，AttributeError会消失，同时可以正常跨两个数据库执行查询操作。注意提前确保测试环境的两个数据库已创建，避免连接失败。

内容的提问来源于stack exchange，提问作者python_help