使用 SqlAlchemy 为单个测试函数重新创建数据库失败

Question

我正在努力更新我对 Flask 和 SqlAlchemy 的了解。我完成了教程 (https://flask.palletsprojects.com/en/2.0.x/tutorial/index.html) 现在我正在调整它以包括 SqlAlchemy。我尽量不使用 Flask-SqlAlchemy 插件，假设这会让我更好地理解事情。

但是，我运行遇到了测试问题。在原始教程中，每个测试函数都创建了一个新数据库。在开始每个测试之前产生一个状态已知的干净数据库。现在我想对 SqlAlchemy 做同样的事情但失败了。请参见下面的代码示例：定义了两个测试（test_create_app 和 test_create_another_app）。当我运行他们分别通过。当我运行他们在一个会话中时，第二个失败。

我可以确认这两个测试都有自己的数据库文件。还可以创建 table 作品。但是我可以看到第二个 table 没有任何用户，而我确实创建了它们。

我是否在尝试做一些我不应该做的事情？或者我在这里做错了什么？如果我查看有关测试数据库的 FastAPI 文档 (https://fastapi.tiangolo.com/advanced/testing-database/)，每个会话只创建一个数据库。我想以某种其他方式管理单个测试的状态。

感谢任何帮助！

import sqlalchemy as sa
from sqlalchemy import create_engine, orm
from sqlalchemy.sql.expression import select
import pytest

session_factory = None

#  ~~~~~~~~~~~~~~~~~~~~~ ORM models

SqlAlchemyBase = sa.ext.declarative.declarative_base()


class User(SqlAlchemyBase):
    __tablename__ = "user"

    id: int = sa.Column(sa.Integer, primary_key=True, autoincrement=True)
    username = sa.Column(sa.String, unique=True, nullable=False)
    password = sa.Column(sa.String, nullable=False)


USERS = [
    User(username="Güllich", password="action direct"),
    User(username="Graham", password="the island"),
]

#  ~~~~~~~~~~~~~~~~~~~~~ Database management


def init_db(connection_url: str):
    """Initialize the database."""
    global session_factory
    if session_factory:
        raise Exception("you should run this function only once.")

    engine = create_engine(connection_url, future=True)
    session_factory = orm.sessionmaker(bind=engine)

    SqlAlchemyBase.metadata.create_all(bind=engine)


def close_db():
    """Close the database"""
    global session_factory
    if session_factory:
        session_factory.close_all()
    session_factory = None


#  ~~~~~~~~~~~~~~~~~~~~~ Testing part.


@pytest.fixture
def dbase(tmp_path):
    sqlite_file = tmp_path / "test_db.sqlite"
    connection_string = f"sqlite+pysqlite:///{sqlite_file.as_posix()}"
    print("creating database at: %s", connection_string)
    init_db(connection_string)

    session = session_factory()

    for user in USERS:
        session.add(user)
    session.commit()
    session.close()

    yield

    close_db()


def test_create_app(dbase):
    db_session = session_factory()
    query = select(User)

    users = db_session.execute(query).scalars().all()

    assert len(users) == 2


def test_create_another_app(dbase):
    db_session = session_factory()
    query = select(User)

    users = db_session.execute(query).scalars().all()

    assert len(users) == 2

Answer 1

我发现我上面的代码有什么问题：

sqlalchemy orm 实例的 USERS 列表 - 我想显然 - 只能绑定到数据库一次。所以第二次尝试这样做会失败（不幸的是默默地）。

使用 SqlAlchemy 为单个测试函数重新创建数据库失败

Recreating a database using SqlAlchemy for individual test functions fails

python

sqlalchemy

flask