Pytest测试SQLAlchemy异步PG删表功能生效但用例报TypeError
核心报错原因
你触发错误的根本原因是delete_table方法中conn.run_sync的调用写法不符合SQLAlchemy异步接口规范:
# 错误写法 await conn.run_sync(Base.metadata.drop_all(sync_engine, [table], checkfirst=True))
conn.run_sync的第一个入参要求是可调用的同步函数对象,但你在入参时直接执行了Base.metadata.drop_all(...),该方法无返回值(返回None),等于你给run_sync传入的执行对象是None,因此抛出TypeError: 'NoneType' object is not callable。
表删除成功是因为drop_all方法在传入run_sync前已经同步执行完成,后续run_sync调用阶段才触发的参数校验错误。
修复方案
调整run_sync的传参方式,将函数对象作为第一个参数,函数需要的参数按顺序跟在后面即可,无需手动传入引擎(run_sync绑定的连接已经关联了对应引擎,会自动注入):
async def delete_table(self, table_name, sync_engine): table = self.meta.tables[table_name] async with self._engine.begin() as conn: # 正确写法:传函数对象+后续参数 await conn.run_sync(Base.metadata.drop_all, [table], checkfirst=True) return
其余可优化点
- 异步测试用例不需要手动调用
asyncio.run,搭配@pytest.mark.asyncio装饰器直接写异步函数即可,避免事件循环冲突:
@pytest.mark.asyncio async def test_drop_table(): await DB.delete_table('table_name') tables = DB.get_table_names() assert 'table_name' not in tables
- 你最初写的fixture
table_names是在测试用例执行前(删表前)获取的表名列表,用这个列表做断言必然失败,删表后重新获取表名列表的写法是正确的。 Db类的meta属性每次访问都会触发一次全库元数据反射,且用了全局sync_engine,建议保持异步/同步引擎的元数据同步,避免出现元数据和实际库结构不一致的问题。
内容的提问来源于stack exchange,提问作者alphaomega
相关产品推荐
相关产品推荐

