SQLAlchemy ORM删表函数主脚本运行正常,Pytest执行报KeyError如何解决?
错误根源
首先明确:KeyError: 'user' 不是PostgreSQL数据库中不存在该表,而是SQLAlchemy本地维护的元数据字典self.meta.tables中没有注册对应表的定义,属于本地缓存未同步的问题,同时你的异步方法写法也存在错误。
修复步骤
1. 修正delete_table异步方法的错误写法
conn.run_sync要求传入可调用对象,而非函数执行结果,你原有写法是直接执行drop_all后将返回值传给run_sync,不符合API要求,同时硬编码Base.metadata、sync_engine会导致元数据不统一,修正后代码:
async def delete_table(self, table_name): # 先反射数据库最新表结构到本地元数据,避免缓存缺失 async with self.engine.begin() as conn: await conn.run_sync(self.meta.reflect, only=[table_name]) table = self.meta.tables[table_name] async with self.engine.begin() as conn: # 正确的run_sync传参:先传函数,后续传函数的关键字参数 await conn.run_sync(self.meta.drop_all, tables=[table], checkfirst=True)
2. 修正测试fixture的实例化逻辑
你原有fixture直接调用类方法DB.delete_table,而delete_table是实例方法,需要和主脚本使用完全相同的初始化逻辑初始化DB实例,确保元数据加载逻辑一致,修改conftest.py代码:
import asyncio import pytest from myproject.src.db import PostgresDb # 导入你实际的DB类 @pytest.fixture(scope="function") def test_db(): # 复用主脚本的DB实例初始化逻辑,保证配置一致 db = PostgresDb() # 初始化阶段反射所有表结构到本地元数据 async def reflect_tables(): async with db.engine.begin() as conn: await conn.run_sync(db.meta.reflect) asyncio.run(reflect_tables()) yield db @pytest.fixture(scope="function") def test_drop_table(test_db): def get_delete_tables(table_name): return asyncio.run(test_db.delete_table(table_name)) return get_delete_tables
3. 修正测试用例的断言逻辑
你原有测试用例中手动raise KeyError会导致无论test_table_commits执行结果如何都会抛出异常,无法正确校验表是否被删除,修改test.py代码:
@pytest.mark.parametrize('function_input, expected', [('user', 'user'), pytest.param('intentional failure', 'intentional failure', marks=pytest.mark.xfail(reason='testing incorrect table name fails'))]) def test_drop_table(test_drop_table, function_input, expected): # 删除表 test_drop_table(function_input) # 校验表不存在时test_table_commits会抛出KeyError with pytest.raises(KeyError) as error_info: test_table_commits(function_input, expected) assert error_info.type is KeyError
4. 可选验证项
如果修改后仍报错,检查pytest运行时的Python导入路径,确保所有ORM模型文件都被正确导入到测试运行环境中,只有模型被导入时才会自动注册到SQLAlchemy的元数据对象中。
内容的提问来源于stack exchange,提问作者alphaomega
相关产品推荐
相关产品推荐

