如何在SQLAlchemy中将CursorResult转换为Pandas DataFrame？

核心解决方案 #

你可以通过CursorResult的内置方法提取数据与元信息，手动映射数据库类型到Pandas dtype，生成指定类型的DataFrame，同时保留Session上下文。

优化后的完整代码 #

from sqlalchemy.orm import Session
from sqlalchemy import create_engine
import pandas as pd
import datetime

engine = create_engine(f"mssql+pyodbc://{db_server}/{db_name}?trusted_connection=yes&driver={db_driver}")
q1 = "SELECT * FROM my_schema.my_table"

with Session(engine) as session:
    results = session.execute(q1)
    # 获取所有数据行
    rows = results.all()
    # 直接提取列名（比cursor.description更简洁）
    colnames = list(results.keys())
    # 获取各列的数据库类型信息
    col_types = [desc[1] for desc in results.cursor.description]

# 构建数据库类型到Pandas dtype的映射（以MSSQL常用类型为例）
dtype_mapping = {}
for col_name, col_type in zip(colnames, col_types):
    if col_type is int:
        dtype_mapping[col_name] = pd.Int64Dtype()  # 支持空值的整数类型
    elif col_type is str:
        dtype_mapping[col_name] = pd.StringDtype()  # 支持空值的字符串类型
    elif col_type is datetime.datetime:
        dtype_mapping[col_name] = pd.DatetimeTZDtype(tz='UTC')  # 按需调整时区
    elif col_type is float:
        dtype_mapping[col_name] = pd.Float64Dtype()
    # 可根据你的实际数据类型扩展更多映射规则

# 创建指定数据类型的DataFrame
df = pd.DataFrame(rows, columns=colnames).astype(dtype_mapping)

关键细节说明 #

1. 简化列名提取：results.keys()直接返回查询结果的列名列表，比从cursor.description提取更简洁可靠。
2. 类型映射逻辑：
   • 从cursor.description拿到的col_type是数据库驱动返回的Python原生类型，需要对应到Pandas的nullable dtype（比如pd.Int64Dtype()而非原生int，避免空值导致的类型报错）。
   • 不同数据库的类型返回可能存在差异，需根据MSSQL实际返回的类型调整映射规则。
3. Session优化：查询操作无需调用session.commit()，只有增删改操作才需要提交事务，可去掉冗余的commit()调用。
4. 便捷替代方案：若在Session上下文内需要执行多批次查询，也可以通过session.bind获取连接，配合pd.read_sql自动处理类型映射，同时保留Session上下文：

   with Session(engine) as session:
       df = pd.read_sql(q1, session.bind)
       # 继续执行其他查询操作...
   

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