使用aiohttp结合asyncpg连接池触发InterfaceError报错求助

抛出的异常：

asyncpg.exceptions._base.InterfaceError: cannot perform operation: another operation is in progress

触发场景：应用收到任意已添加路由的GET请求时都会抛出该异常，即使使用PgBouncer也无法解决。

原代码 #

import asyncio
import asyncpg
from aiohttp import web


async def handle(request):
    """Handle incoming requests."""
    pool = request.app['pool']
    power = int(request.match_info.get('power', 10))

    # Take a connection from the pool.
    async with pool.acquire() as connection:
        # Open a transaction.
        async with connection.transaction():
            # Run the query passing the request argument.
            result = await connection.fetchval('select 2 ^ $1', power)
            return web.Response(
                text="2 ^ {} is {}".format(power, result))


async def init_app():
    """Initialize the application server."""
    app = web.Application()
    # Create a database connection pool
    app['pool'] = await asyncpg.create_pool(
        host="DB_HOST",
        port=5432,
        user="DB_USER",
        password="DB_PASSWORD",
        database="DB_NAME",)
    # Configure service routes
    app.router.add_route('GET', '/issues', handle)
    app.router.add_route('GET', '/', handle)
    return app


loop = asyncio.get_event_loop()
app = loop.run_until_complete(init_app())
web.run_app(app, port=8800)

问题原因与修复 #

问题出在事务上下文未完成时就提前return，导致连接被归还到池中的时候，事务还处于未结束的状态，后续复用该连接时就会触发"另一个操作正在进行"的错误。

修复方法很简单：把return语句移到事务和连接的上下文外面，确保事务正常提交/回滚，连接彻底释放后再返回响应：

修改后的代码 #

import asyncio
import asyncpg
from aiohttp import web


async def handle(request):
    """Handle incoming requests."""
    pool = request.app['pool']
    power = int(request.match_info.get('power', 10))
    result = None

    # Take a connection from the pool.
    async with pool.acquire() as connection:
        # Open a transaction.
        async with connection.transaction():
            # Run the query passing the request argument.
            result = await connection.fetchval('select 2 ^ $1', power)
    # 事务和连接上下文已退出，再返回响应
    return web.Response(
        text="2 ^ {} is {}".format(power, result))


async def init_app():
    """Initialize the application server."""
    app = web.Application()
    # Create a database connection pool
    app['pool'] = await asyncpg.create_pool(
        host="DB_HOST",
        port=5432,
        user="DB_USER",
        password="DB_PASSWORD",
        database="DB_NAME",)
    # Configure service routes
    app.router.add_route('GET', '/issues', handle)
    app.router.add_route('GET', '/', handle)
    return app


loop = asyncio.get_event_loop()
app = loop.run_until_complete(init_app())
web.run_app(app, port=8800)

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