Flask/FastAPI SQLite pytest fixture returns None 除非指定行 ID

Flask/FastAPI SQLite pytest fixture returns None unless row ID is specified

我正在使用 pytest 测试 FastAPI 应用程序。我创建了一个客户端装置,其中包括一个从 CSV 创建的 sqlite 数据库:

import pytest
from os import path, listdir, remove
from pandas import read_csv
from fastapi.testclient import TestClient
from api.main import app
from api.db import engine, db_url

@pytest.fixture(scope="session")
def client():
    db_path = db_url.split("///")[-1]
    if path.exists(db_path):
        remove(db_path)
    file_path = path.dirname(path.realpath(__file__))
    table_path = path.join(file_path, "mockdb")
    for table in listdir(table_path):
        df = read_csv(path.join(table_path, table))
        df.to_sql(table.split('.')[0], engine, if_exists="append", index=False)
    client = TestClient(app)
    yield client

我在 FastAPI 应用程序中的数据库设置:

import os
from sqlalchemy import create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

dirname = os.path.dirname(__file__)
if "pytest" in modules:
    mock_db_path = os.path.join(dirname, '../test/mockdb/test.db')
    db_url = f"sqlite:///{mock_db_path}"
else:
    db_url = os.environ.get("DATABASE_URL", None)
if "sqlite" in db_url:
    engine = create_engine(db_url, connect_args={"check_same_thread": False})
else:
    engine = create_engine(db_url)
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
Base = declarative_base()

这行得通:我可以为查询数据库的应用端点设置测试,并返回我放入 CSV 中的数据,例如在 mockdb/person.csv 添加一行后:

from api.db import SessionLocal

db = SessionLocal()
all = db.query(Person).all()
print(all)
[<tables.Person object at 0x7fc829f81430>]

我现在正在尝试测试向数据库中的表添加新行的代码。

这仅在我指定 ID 时有效(假设这发生在 pytest 运行 期间):

db.add(Person(id=2, name="Alice"))
db.commit()
all = db.query(Person).all()
print(all)
[<tables.Person object at 0x7fc829f81430>, <tables.Person object at 0x7fc829f3bdc0>]

以上结果与我预期的程序行为一致。但是,如果我不指定 ID,则结果为 None:

db.add(Person(name="Alice"))
db.commit()
all = db.query(Person).all()
print(all)
[<tables.Person object at 0x7fc829f81430>, None]

这个结果不是我期望的程序行为方式。

我要测试的代码没有指定 ID,它使用自动增量作为一种好的做法。因此,我无法测试这段代码。它只是创建这些 Nones.

起初,我认为罪魁祸首不是使用 Base.metadata.create_all() 创建表。但是,我已经尝试将它放在我的客户端装置中,并遵循我的数据库设置(即上面的前 2 个代码块),但结果是相同的:Nones.

使用调试器单步执行,添加 Person 行时,出现以下错误:

sqlalchemy.orm.exc.ObjectDeletedError: Instance '<Person at 0x7fc829f3bdc0>' has been deleted, or its row is otherwise not present.

为什么结果行是 None,我该如何解决这个错误?

错误的原因是我的数据库中有一个与 SQLite 不兼容的列类型,即 PostgresSQL 的 ARRAY 类型。不幸的是,没有错误消息提示这一点。最简单的解决方案是删除或更改此列的类型。

也可以通过如下更改client()来保留列和 SQLite fixture:

from mytableschema import MyOffendingTable

@pytest.fixture(scope="session")
def client():
    table_meta = SBEvent.metadata.tables[MyOffendingTable.__tablename__]
    table_meta._columns.remove(table_meta._columns["my_offending_column"])
    Base.metadata.create_all(bind=engine)
    db_path = db_url.split("///")[-1]
    if path.exists(db_path):
        remove(db_path)
    file_path = path.dirname(path.realpath(__file__))
    table_path = path.join(file_path, "mockdb")
    for table in listdir(table_path):
        df = read_csv(path.join(table_path, table))
        df.to_sql(table.split('.')[0], engine, if_exists="append", index=False)
    client = TestClient(app)
    yield client

如果您从 MyOffendingTable CSV 中删除 my_offending_column,现在可以正常进行。没有更多 Nones!

可悲的是,在测试 运行 期间查询有问题的 table 仍然会 运行 出现问题,因为 SELECT 语句将查找不存在的 my_offending_column。需要查询said table的,推荐使用dialect-specific compilation rules.