时间戳

首页 > 解决方案 > Flask/FastAPI SQLite pytest 夹具返回 None 除非指定行 ID

python - Flask/FastAPI SQLite pytest 夹具返回 None 除非指定行 ID

问题描述

我正在用 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 个代码块），但结果是相同None的：

使用调试器单步执行，添加 Person 行时，会出现以下错误：

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

为什么结果行None以及如何解决此错误？

标签： pythonsqlalchemypytestfastapi

解决方案

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

也可以通过client()如下更改来保留列和 SQLite 夹具：

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

my_offending_column如果您从MyOffendingTableCSV中删除，现在可以正常进行。没有了None！

遗憾的是，在测试运行期间查询有问题的表仍然会遇到问题，因为 SELECT 语句将查找不存在的my_offending_column. 对于那些需要查询该表的人，我建议使用特定方言的编译规则。

推荐阅读