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,它使用自动增量作为一种好的做法。因此,我无法测试这段代码。它只是创建这些 None
s.
起初,我认为罪魁祸首不是使用 Base.metadata.create_all()
创建表。但是,我已经尝试将它放在我的客户端装置中,并遵循我的数据库设置(即上面的前 2 个代码块),但结果是相同的:None
s.
使用调试器单步执行,添加 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
,现在可以正常进行。没有更多 None
s!
可悲的是,在测试 运行 期间查询有问题的 table 仍然会 运行 出现问题,因为 SELECT 语句将查找不存在的 my_offending_column
。需要查询said table的,推荐使用dialect-specific compilation rules.
我正在使用 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,它使用自动增量作为一种好的做法。因此,我无法测试这段代码。它只是创建这些 None
s.
起初,我认为罪魁祸首不是使用 Base.metadata.create_all()
创建表。但是,我已经尝试将它放在我的客户端装置中,并遵循我的数据库设置(即上面的前 2 个代码块),但结果是相同的:None
s.
使用调试器单步执行,添加 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
,现在可以正常进行。没有更多 None
s!
可悲的是,在测试 运行 期间查询有问题的 table 仍然会 运行 出现问题,因为 SELECT 语句将查找不存在的 my_offending_column
。需要查询said table的,推荐使用dialect-specific compilation rules.