如何在Async FastAPI+SQLAlchemy中将数据库数据转为Pandas DataFrame？

问题根源 #

Pandas的pd.read_sql目前不支持SQLAlchemy的异步引擎（AsyncEngine），它依赖同步数据库连接的cursor接口，而异步引擎没有这个属性，因此抛出AttributeError: 'AsyncEngine' object has no attribute 'cursor'。另外你调用异步函数时未加await，会导致返回协程对象而非实际查询结果。

解决方法 #

方法一：异步查询后转换为DataFrame #

先通过SQLAlchemy异步Session执行查询，获取结果后手动转换为Pandas DataFrame，这是最贴合异步环境的方案：

async def get_all_exercises(session):
    # 异步执行查询
    result = await session.execute(select(models.Exercises))
    # 获取所有实体对象
    exercises = result.scalars().all()
    # 将实体对象转为字典列表，再生成DataFrame
    df = pd.DataFrame([item.__dict__ for item in exercises])
    # 移除SQLAlchemy内部维护的属性
    df = df.drop(columns=["_sa_instance_state"])
    print(type(df))
    return df

# 调用时需添加await
async with async_session() as session:
    async with session.begin():
        df = await get_all_exercises(session)
        print(df)

方法二：单独创建同步引擎供Pandas使用 #

如果坚持要用pd.read_sql，可以单独创建一个SQLAlchemy同步引擎（与异步引擎共存），注意要替换掉URL中的异步驱动后缀（比如+asyncpg改为普通的PostgreSQL驱动）：

# 新增同步引擎示例（以PostgreSQL为例）
from sqlalchemy import create_engine

# 替换异步驱动，比如原URL是"postgresql+asyncpg://user:pass@host/db"，去掉+asyncpg
SYNC_DATABASE_URL = DATABASE_URL.replace("+asyncpg", "")
sync_engine = create_engine(SYNC_DATABASE_URL, future=True, echo=False)

async def get_all_exercises():
    # 直接用同步引擎调用read_sql
    df = pd.read_sql(select(models.Exercises), sync_engine)
    print(type(df))
    return df

# 调用示例
async with async_session() as session:
    async with session.begin():
        df = await get_all_exercises()
        print(df)

注意事项 #

• 异步函数必须通过await调用，否则只会返回协程对象，无法获取实际数据。
• 如果使用方法二，确保数据库允许同时建立同步和异步连接，且驱动依赖已安装（比如同步PostgreSQL需要psycopg2-binary）。

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