SQLAlchemy 模拟会话连接答案

【问题标题】：SQLAlchemy mock Session connectionSQLAlchemy 模拟会话连接
【发布时间】：2016-12-25 14:32:02
【问题描述】：

在我需要在应用程序中查询数据库的任何地方，我都会从这个文件中导入 Session。

from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import scoped_session, sessionmaker

db_name = 'analytics'
db_url = 'postgresql+psycopg2://**:**@localhost:5432/{0}'.format(db_name)
Engine = create_engine(db_url, echo=False)
SessionMaker = sessionmaker(bind=Engine, autoflush=False)
Session = scoped_session(SessionMaker)

Base = declarative_base()

我正在编写一些单元测试并且需要模拟 Session 对象，以便每当它被导入我的应用程序时，它都会连接到 unittest 数据库而不是分析数据库。

class FacebookTest(unittest.TestCase):

    @classmethod
    def setUpClass(self):
        """
        create unittest database and insert Premier League 2015-2016 season
        """
        con = connect(user='**', host='localhost', password='**', database='postgres')
        con.set_isolation_level(ISOLATION_LEVEL_AUTOCOMMIT)
        cur = con.cursor()
        cur.execute('DROP DATABASE IF EXISTS unittest')
        cur.execute('CREATE DATABASE unittest')
        cur.close()
        con.close()
        Engine = create_engine('postgresql+psycopg2://**:**@localhost:5432/unittest')
        Base.metadata.create_all(Engine)
        Session.connection = Engine.connect()

【问题讨论】：

你有什么问题？
如何模拟会话，以便在运行测试时连接到 unittest 数据库。

标签： python session sqlalchemy

【解决方案1】：

在不同的端口上设置您的模拟数据库，例如 5433，并根据您要连接的数据库更改端口。

你可以这样做。假设您有一个 db.py，您可以在其中启动会话并将它们全部导入。像这样修改它（忽略对配置器的引用，这是我的内部模块）：

import psycopg2
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
import configurator

Base = declarative_base()

def connect():
    cfg = configurator.Configurator("database")
    return psycopg2.connect(
        user=cfg.user, host=cfg.host, dbname=cfg.db, port=cfg.port, password=cfg.pwd
    )

def connect_test():
    conn_string = (
        "host='localhost' user='postgres' password='yourPassword' port='5433'"
    )
    return psycopg2.connect(conn_string)

def Session(database='dev'):
    if database == 'dev':
        engine = create_engine("postgresql://", creator=connect)
    elif database == 'test':
        engine = create_engine("postgresql://", creator=connect_test)
    Session = sessionmaker(bind=engine)
    return Session()

更新：重要的更正 - 在 return session() 中添加了括号，对此感到抱歉。

UPDATE2：显式测试后更改了代码。

【讨论】：