FastAPI+SQLAlchemy+asyncpg切换PostgreSQL后请求报错：存在正在进行的操作

错误原因分析 #

这个错误是因为同一个asyncpg连接上同时存在多个未完成的数据库操作。sqlite+aiosqlite未触发问题，是因为它的异步实现基于单线程，天然不会有并发操作冲突；而asyncpg是真正的异步驱动，会严格校验连接的操作状态，会话管理不当就会触发该错误。

你的代码核心问题是异步生成器的类型标注错误，FastAPI依赖注入无法正确识别异步上下文的生命周期，导致会话未被正确释放或复用，进而引发连接上的并发操作冲突。

解决方案 #

1. 修正依赖函数的类型标注 #

将Generator替换为AsyncGenerator（需从typing导入），这是异步生成器的正确类型：

from typing import AsyncGenerator  # 新增导入
import os

from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
from sqlalchemy.orm import sessionmaker

# 数据库URL和引擎创建部分保持不变
user = os.getenv('PG_USER')
password = os.getenv('PG_PASSWORD')
domain = os.getenv('PG_DOMAIN')
db = os.getenv('PG_DATABASE')

db_async_url = f'postgresql+asyncpg://{user}:{password}@{domain}/{db}'

async_engine = create_async_engine(
    db_async_url, future=True, echo=True
)

create_async_session = sessionmaker(
    async_engine, class_=AsyncSession, expire_on_commit=False
)

# 修正后的依赖函数
async def get_async_session() -> AsyncGenerator[AsyncSession, None]:
    async with create_async_session() as session:
        yield session

2. 确保所有数据库操作都使用await #

SQLAlchemy异步API的所有数据库操作必须用await调用，禁止使用同步写法：

错误写法（同步）：

# 未await会导致操作占用连接不释放
users = session.query(User).all()

正确写法（异步）：

from sqlalchemy import select

# 查询多条数据
users = await session.scalars(select(User)).all()
# 查询单条数据
user = await session.get(User, user_id)
# 提交事务
await session.commit()

3. 避免同一会话内并发操作 #

如果需要并行执行多个数据库操作，要为每个操作分配独立会话，不要在同一个会话里用asyncio.gather执行多数据库任务。

验证修正 #

修改后重启FastAPI服务，测试请求应不再出现InterfaceError。若仍有问题，可检查：

• 环境变量是否正确配置，确保连接目标PostgreSQL实例
• SQLAlchemy版本是否为1.4+（异步API需该版本及以上）
• 是否有其他地方手动创建会话但未正确释放

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