使用Pandas+SQLAlchemy向SQL Server插入数据时遇fetchall属性错误

报错信息 #

Traceback (most recent call last):
File "C:\Users\neyma\PycharmProjects\pythonProject23\main.py", line 95, in
for row in conn.fetchall():
AttributeError: 'Connection' object has no attribute 'fetchall'

你的代码 #

connection_url = URL.create(
    "mssql+pyodbc",
    query={"odbc_connect": connection_string}
)
engine = create_engine(connection_url)
engine = create_engine(connection_url)  # 重复创建engine
conn = engine.connect().execution_options(
    stream_results=True)


with open("ss.xml") as fp:
    soup = BeautifulSoup(fp, 'xml')
data = []

# 注：原代码缺失循环逻辑，此处补充遍历Event节点的假设
for Event in soup.find_all('Event'):
    data.append({
        'DSName': e.text if (e := Event.select_one(('Data[Name="DSName"]'))) else None,
        'DSType': e.text if (e := Event.select_one(('Data[Name="DSType"]'))) else None,
        'ObjectDN': e.text if (e := Event.select_one(('Data[Name="ObjectDN"]'))) else None,
        'ObjectGUID': e.text if (e := Event.select_one(('Data[Name="ObjectGUID"]'))) else None,
        'ObjectClass': e.text if (e := Event.select_one(('Data[Name="ObjectClass"]'))) else None,
        'AttributeLDAPDisplayName': e.text if (e := Event.select_one(('Data[Name="AttributeLDAPDisplayName"]'))) else None,
        'AttributeSyntaxOID': e.text if (e := Event.select_one(('Data[Name="AttributeSyntaxOID"]'))) else None,
        'AttributeValue': e.text if (e := Event.select_one(('Data[Name="AttributeValue"]'))) else None,
        'OperationType': e.text if (e := Event.select_one(('Data[Name="OperationType"]'))) else None,
    })

df = pd.DataFrame(data);
engine.execute('''
                            CREATE TABLE try(
                                DSName nvarchar(max),
                                DSType nvarchar(max),
                                ObjectDN nvarchar(max),
                                ObjectGUID nvarchar (max),
                                AttributeLDAPDisplayName nvarchar(max),
                                AttributeSyntaxOID nvarchar(max),
                                AttributeValue nvarchar(max),
                                OperationType nvarchar(max),  # 此处多了逗号，会触发SQL语法错误
                                )
                                   ''')
df.to_sql('try', conn, if_exists='replace', index = False,chunksize=100)
engine.execute('''  
SELECT * FROM test  # 表名与插入的try不一致
          ''')

for row in conn.fetchall():  # 错误：conn是Connection对象，无fetchall方法
    print (row)

问题解决 #

1. 核心错误原因 #

conn是SQLAlchemy的Connection对象，fetchall()是查询结果集（ResultProxy）的方法，不是连接对象的方法。必须先执行查询并保存结果集，再调用fetchall()。

2. 完整修正步骤 #

• 移除重复创建的engine实例
• 修正CREATE TABLE的语法错误（去掉最后一个字段后的逗号）
• 统一表名（插入和查询都使用try）
• 执行查询时保存结果集，再调用fetchall()
• 可选优化：df.to_sql可直接传入engine，无需单独的conn对象

修正后的代码 #

from sqlalchemy import create_engine, URL
from bs4 import BeautifulSoup
import pandas as pd

connection_url = URL.create(
    "mssql+pyodbc",
    query={"odbc_connect": connection_string}
)
engine = create_engine(connection_url)

with open("ss.xml") as fp:
    soup = BeautifulSoup(fp, 'xml')
data = []

# 遍历XML中的Event节点提取数据
for Event in soup.find_all('Event'):
    data.append({
        'DSName': e.text if (e := Event.select_one('Data[Name="DSName"]')) else None,
        'DSType': e.text if (e := Event.select_one('Data[Name="DSType"]')) else None,
        'ObjectDN': e.text if (e := Event.select_one('Data[Name="ObjectDN"]')) else None,
        'ObjectGUID': e.text if (e := Event.select_one('Data[Name="ObjectGUID"]')) else None,
        'ObjectClass': e.text if (e := Event.select_one('Data[Name="ObjectClass"]')) else None,
        'AttributeLDAPDisplayName': e.text if (e := Event.select_one('Data[Name="AttributeLDAPDisplayName"]')) else None,
        'AttributeSyntaxOID': e.text if (e := Event.select_one('Data[Name="AttributeSyntaxOID"]')) else None,
        'AttributeValue': e.text if (e := Event.select_one('Data[Name="AttributeValue"]')) else None,
        'OperationType': e.text if (e := Event.select_one('Data[Name="OperationType"]')) else None,
    })

df = pd.DataFrame(data)

# 修正CREATE TABLE语法错误
engine.execute('''
CREATE TABLE try(
    DSName nvarchar(max),
    DSType nvarchar(max),
    ObjectDN nvarchar(max),
    ObjectGUID nvarchar(max),
    AttributeLDAPDisplayName nvarchar(max),
    AttributeSyntaxOID nvarchar(max),
    AttributeValue nvarchar(max),
    OperationType nvarchar(max)
)
''')

# 直接使用engine传入to_sql，无需单独conn
df.to_sql('try', engine, if_exists='replace', index=False, chunksize=100)

# 执行查询并保存结果集
result = engine.execute('SELECT * FROM try')
# 从结果集调用fetchall()遍历数据
for row in result.fetchall():
    print(row)

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