You need to enable JavaScript to run this app.
优惠活动
大模型
产品
解决方案
定价
更多

Python执行无返回结果的SQL查询时如何判断是否运行成功?

执行无结果集SQL的成功判定方法
  • 移除所有fetchone()/fetchall()/fetchmany()调用:你遇到的pyodbc.ProgrammingError: No results完全是调用读取结果方法导致的,和SQL本身执行成功与否无关,这类SELECT INTO操作本身就没有结果集返回,不需要调用任何fetch类方法。
  • 用异常捕获+受影响行数双重校验:
    1. 用try-except包裹cursor.execute()执行逻辑,所有SQL执行错误(语法错误、权限不足、表已存在等)都会以pyodbc.Error子类异常的形式抛出,进入异常分支即可判定执行失败。你之前的报错被fetch调用的异常掩盖,移除fetch代码后就能拿到SQL本身的具体错误信息。
    2. 执行无异常时,读取cursor.rowcount属性,该值和SSMS返回的「rows affected」行数完全一致,和你的预期行数匹配即可确认执行成功。
  • 不要遗漏事务提交:pyodbc默认关闭自动提交,你可以在创建连接时指定autocommit=True,或者执行完SQL后手动调用conn.commit(),否则即使执行无报错,操作也不会真正写入数据库。

参考实现代码

import pyodbc

# 数据库连接配置,根据实际情况修改
conn_str = "DRIVER={ODBC Driver 17 for SQL Server};SERVER=你的服务器地址;DATABASE=你的库名;UID=用户名;PWD=密码"
# 开启自动提交就不需要后续手动commit
conn = pyodbc.connect(conn_str, autocommit=True)
cursor = conn.cursor()

create_table_sql = """
SELECT *
INTO MyDataBase.MySchema.NewTable
FROM
    (SELECT
        ID,
        Count(*) AS ProductCount
    FROM
        MyDataBase.MySchema.MyTable
    GROUP BY 
        ID
    ) ProductCount
"""

try:
    cursor.execute(create_table_sql)
    print(f"执行成功,共写入{cursor.rowcount}行数据")
    # 若需要更高的校验级别,可额外执行查询确认表存在、行数匹配
    # cursor.execute("SELECT COUNT(*) FROM MyDataBase.MySchema.NewTable")
    # print(f"新表实际行数:{cursor.fetchone()[0]}")
except pyodbc.Error as e:
    print(f"执行失败,错误详情:{str(e)}")
finally:
    cursor.close()
    conn.close()

内容的提问来源于stack exchange,提问作者Maxcot

相关产品推荐
方舟 Agent Plan

超全模态模型 × Harness 升级,最新支持 Deepseek-V4.1-Flash、GLM-5.3 系列、Doubao-Seedream-5.0-pro、Kimi-K3 (部分), 限时 9.9 元起

最近更新时间:2026.09.23 19:15:02