如何在查询时在模式之间切换

Question

通过 MRE，我有一个名为 query 的简单模块，它向 Bet 表添加一条记录：

from container_test import Bet, DataAccessLayer

def main():
    dal = DataAccessLayer()
    dal.create_session()
    query_bets(dal)

def add_new_id(dal):
    new_id = {"id_": 3}
    dal.session.add(Bet(**new_id))
    dal.session.commit()

if __name__ == "__main__":
    main()

container_test 包含以下内容：

from sqlalchemy import Column, Integer, create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class Bet(Base):

    __tablename__ = "bet"
    __table_args__ = ({"schema": "belgarath", "extend_existing": True})

    id_ = Column(Integer, primary_key=True)

class DataAccessLayer():

    def __init__(self):
        conn_string = "mysql+mysqlconnector://root:root@localhost/"
        self.engine = create_engine(conn_string)

    def create_session(self):
        Base.metadata.create_all(self.engine)
        Session = sessionmaker()
        Session.configure(bind=self.engine)
        self.session = Session()

我希望能够测试query_bets 函数，为此我在另一个文件中创建了一个简单的pytest 函数：

def test_add_new_id(
    create_dal, # creates an instance of a dal via conftest
    setup_teardown, # creates a blank belgarath_test schema and deletes afterwards via conftest
):
    expected = 3
    sandpit.add_new_id(create_dal)
    actual = query_bet(create_dal) # function that will return a scalar result
    assert actual == expected

但是，我希望能够将标准 belgarath 架构更改为测试架构； belgarath_test 所以我不会冒主数据库完整性的风险。我想我需要某种可以在test_add_new_id 测试功能中激活的“开关”，但我不确定如何创建。

我一直在尝试遵循这篇文章中提出的解决方案：How do I change the schema for both a table and a foreign key? 但是我很挣扎。我不明白我需要在 test_add_new_id 测试函数中做什么来更改架构。

作为参考，我根据解决方案创建了以下模块：

container_test:

from sqlalchemy import Column, Integer, create_engine
from sqlalchemy.orm import sessionmaker

from base_class import SCHEMA_TEST, Base, declared_attr


class Bet(Base):

    @declared_attr
    def __table_args__(cls):
        return {'schema': SCHEMA_TEST}

    __tablename__ = "bet"

    id_ = Column(Integer, primary_key=True)


class DataAccessLayer():

    def __init__(self):
        conn_string = "mysql+mysqlconnector://root:root@localhost/"
        self.engine = create_engine(conn_string)

    def create_session(self):
        Base.metadata.create_all(self.engine)
        Session = sessionmaker()
        Session.configure(bind=self.engine)
        self.session = Session()

base_class（尝试使用__init__.py时出错）：

from sqlalchemy.ext.declarative import declarative_base, declared_attr

SCHEMA_MAIN = 'belgarath'  # figure out how you want to retrieve this
SCHEMA_TEST = 'belgarath_test'


class _Base():

    @declared_attr
    def __table_args__(cls):
        return {'schema': SCHEMA_MAIN}


Base = declarative_base(cls=_Base)
Base.metadata.schema = SCHEMA_MAIN

这成功地将belgarath 更改为belgarath_test，但我只希望在运行test_add_new_id 测试函数时发生这种情况。

Answer 1

与其在同一个数据库的不同模式中运行测试，不如在不同的数据库中运行测试？

这也将改善测试隔离，因为测试将不再有生产数据库的密码，因此有缺陷的测试将无法破坏产品数据。

Answer 2

从 sqlalchemy 1.1 开始，您可以使用以下命令在运行时轻松更改架构：schema_translation_map。

正如在cmets中讨论过的那样，单个会话需要使用4个模式，不建议重复创建会话而简单地使用：

session.connection(execution_options={"schema_translate_map": {"belgarath": "belgarath_test"}})

上述语句会将架构切换为belgarath_test，同样可以使用oncourt和oncourt_test。