在Azure Functions中使用psycopg3异步连接池时如何修改asyncio事件循环策略？

问题描述 #

用Azure Functions + Python连接PostgreSQL，打算通过psycopg3的AsyncConnectionPool实现异步连接来提升性能。按照psycopg要求设置了Windows下的asyncio事件循环策略（asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())），直接运行代码时一切正常，但通过func host start启动本地服务后发起请求就报错，提示仍在使用不兼容的ProactorEventLoop。问题根源在于Azure Functions运行时在我们的设置代码执行前，已经提前初始化了事件循环。

复现代码 #

# function_app.py
import azure.functions as func
import psycopg_pool
import json
import asyncio
import os

if os.name == 'nt':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

con_str = "host=localhost dbname=test-db user=postgres password=admin"
apool = psycopg_pool.AsyncConnectionPool(
            conninfo=con_str,
            open = False
        )

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

async def connect():
    await apool.open()
    await apool.wait()
    print("Connect OK")

@app.route(route="info", auth_level= func.AuthLevel.FUNCTION)
async def info_trigger(req: func.HttpRequest) -> func.HttpResponse:
    try:
        await connect()
        status = {"test": "OK"}
    except:
        status = {"test": "Not ok"}
    return func.HttpResponse(
        json.dumps(status),
        status_code=200,
        mimetype='application/json'
    )

if __name__ == "__main__":
    asyncio.run(connect())

报错信息 #

error connecting in 'pool-1': Psycopg cannot use the 'ProactorEventLoop' to run in async mode. Please use a compatible event loop, for instance by setting 'asyncio.set_event_loop_policy(WindowsSelectorEventLoopPolicy())'

解决方案 #

核心修改：提前设置事件循环策略 #

把事件循环策略的设置代码放在文件最顶部，确保在Azure Functions运行时初始化事件循环前执行，也就是在导入azure.functions等模块之前完成设置：

# function_app.py
# 最优先执行事件循环策略设置
import asyncio
import os

if os.name == 'nt':
    asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

# 再导入其他依赖模块
import azure.functions as func
import psycopg_pool
import json

con_str = "host=localhost dbname=test-db user=postgres password=admin"
apool = psycopg_pool.AsyncConnectionPool(
            conninfo=con_str,
            open = False
        )

app = func.FunctionApp(http_auth_level=func.AuthLevel.FUNCTION)

async def connect():
    await apool.open()
    await apool.wait()
    print("Connect OK")

@app.route(route="info", auth_level= func.AuthLevel.FUNCTION)
async def info_trigger(req: func.HttpRequest) -> func.HttpResponse:
    try:
        await connect()
        status = {"test": "OK"}
    except Exception as e:
        # 新增异常打印，方便排查问题
        print(f"异常信息: {str(e)}")
        status = {"test": "Not ok"}
    return func.HttpResponse(
        json.dumps(status),
        status_code=200,
        mimetype='application/json'
    )

if __name__ == "__main__":
    asyncio.run(connect())

补充说明 #

• 该问题仅出现在Windows本地开发环境，Linux环境下asyncio默认事件循环与psycopg3兼容，无需额外设置。
• 若部署到Azure云环境（Linux容器），直接移除Windows相关的事件循环设置代码即可。

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