如何使用SQLAlchemy执行SQL查询并将结果导入pandas DataFrame

报错根因 #

你遇到的AttributeError: 'ResultProxy' object has no attribute 'all'报错是SQLAlchemy版本差异导致的：1.4版本以前的SQLAlchemy查询返回ResultProxy对象，获取全量结果的方法为fetchall()，all()是1.4及以上版本新增的Result对象的方法。

修复同步查询代码 #

from sqlalchemy import text

with engine.connect() as conn:
    result = conn.execute(text("select * FROM users"))
    # 低版本SQLAlchemy替换为fetchall()即可
    print(result.fetchall())

查询结果转pandas DataFrame #

方式1：同步查询直接转换 #

pandas原生支持从SQLAlchemy连接读取数据，无需手动解析查询结果：

import pandas as pd
from sqlalchemy import text

with engine.connect() as conn:
    df = pd.read_sql_query(text("SELECT * FROM users"), conn)
print(df.head())

方式2：适配异步环境的查询转换 #

你当前使用的是FastAPI异步架构+databases异步连接池，建议优先用异步查询避免阻塞事件循环，直接复用你已经初始化的database对象即可：

import pandas as pd
from sqlalchemy import text

# 可直接写在FastAPI异步路由中
@app.get("/query-users")
async def query_users():
    query = text("SELECT * FROM users")
    # 异步执行查询
    result = await database.fetch_all(query)
    # 转换为DataFrame
    df = pd.DataFrame([dict(row) for row in result])
    # 按需处理df后返回即可
    return {"total": len(df), "data": df.to_dict(orient="records")}

可选优化 #

如果需要使用all()方法，可将SQLAlchemy升级到1.4及以上版本，升级前注意核对已有代码的API兼容性。

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