小编Pro*_*leg的帖子

我想测试一些与asyncpg. 如果我一次运行一个测试，效果很好。但是，如果我一次运行多个测试，则除第一个测试之外的所有测试都会因错误而崩溃asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress。

测试：

@pytest.mark.asyncio\nasync def test_project_connection(superuser_id, project_id):\n    data = element_data_random(project_id)\n\n    element_id = (await resolve_element_create(data=data, user_id=superuser_id))["id"]\n    project_elements = (await db_projects_element_ids_get([project_id]))[project_id]\n\n    assert element_id in project_elements\n\n\n@pytest.mark.asyncio\nasync def test_project_does_not_exist(superuser_id):\n    data = element_data_random(str(uuid.uuid4()))\n\n    with pytest.raises(ObjectWithIdDoesNotExistError):\n        await resolve_element_create(data=data, user_id=superuser_id)\n

使用数据库使用池的所有函数如下所示：

async def <some_db_func>(*args):\n    pool = await get_pool()\n\n    await pool.execute(...) # or fetch/fetchrow/fetchval\n

我如何获得池：

db_pool = None\n\n\nasync def get_pool():\n    global db_pool\n\n    async def init(con):\n        await con.set_type_codec('jsonb', encoder=ujson.dumps, decoder=ujson.loads, schema='pg_catalog')\n …