使用pytest-asyncio搭配aiopg时遭遇NotImplementedError问题求助

我正在使用FastAPI和aiopg开发小型教育类应用，目前仅剩测试代码待编写。尝试通过pytest-asyncio结合依赖覆盖连接测试数据库，相关代码如下：

测试Fixture代码 #

@pytest.fixture(scope="session", autouse=True)
def setup_test_db():
    create_db()
    
    sql_script_path = "./db.sql"
    try:
        conn = ps.connect(DSN_TEST)
        cursor = conn.cursor()
        with open(sql_script_path, "r") as script_file:
            sql_script = script_file.read()
            cursor.execute(sql_script)
        conn.commit()
        conn.close()
    except ps.Error as e:
        raise Exception(f"Error initializing the test database: {e}")

    app.dependency_overrides[get_database_pool] = get_test_database_pool
    app.dependency_overrides[get_db] = get_test_db
    
    yield

    delete_db()

@pytest.fixture
async def ac():
    async with AsyncClient(app=app, base_url="http://test") as ac:
        yield ac

测试用例代码 #

async def test_root(ac: AsyncClient):
    response = await ac.get("/orders/")
    assert response.status_code == 200

已在pytest.ini中设置asyncio_mode = True，运行pytest时出现如下错误：

pool = await aiopg.create_pool(DSN_TEST)  # type: ignore
....\Lib\site-packages\aiopg\pool.py:300: in from_pool_fill
await self._fill_free_pool(False)
....\Lib\site-packages\aiopg\pool.py:336: in _fill_free_pool
conn = await connect(
....\Lib\site-packages\aiopg\connection.py:65: in connect
connection = Connection(
....\Lib\site-packages\aiopg\connection.py:772: in __init__
self._loop.add_reader(

_______________________________________________________________________

self = <ProactorEventLoop running=False closed=False debug=False>, fd = 884, callback = <function Connection._ready at 0x0000011210F49080>  
args = (<weakref at 0x000001121331D080; to 'Connection' at 0x0000011213327990>,)

    def add_reader(self, fd, callback, *args):

>       raise NotImplementedError

E       NotImplementedError

........\AppData\Local\Programs\Python\Python311\Lib\asyncio\events.py:530: NotImplementedError

解决方法 #

问题根源 #

Windows系统下Python默认使用ProactorEventLoop，而该事件循环不支持add_reader方法，aiopg内部依赖此方法建立数据库连接，因此触发报错。

具体解决方案 #

在测试文件的顶部（或项目的conftest.py中）添加事件循环策略配置，强制使用支持add_reader的SelectorEventLoop：

import asyncio

# 针对Windows系统设置事件循环策略
asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

添加后重新运行pytest，即可解决该错误。

内容的提问来源于stack exchange，提问作者Ririol