You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

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
  • 你最初写的fixturetable_names是在测试用例执行前(删表前)获取的表名列表,用这个列表做断言必然失败,删表后重新获取表名列表的写法是正确的。
  • Db类的meta属性每次访问都会触发一次全库元数据反射,且用了全局sync_engine,建议保持异步/同步引擎的元数据同步,避免出现元数据和实际库结构不一致的问题。

内容的提问来源于stack exchange,提问作者alphaomega

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.23 18:36:01