Python执行无返回结果的SQL查询时如何判断是否运行成功?
执行无结果集SQL的成功判定方法
- 移除所有
fetchone()/fetchall()/fetchmany()调用:你遇到的pyodbc.ProgrammingError: No results完全是调用读取结果方法导致的,和SQL本身执行成功与否无关,这类SELECT INTO操作本身就没有结果集返回,不需要调用任何fetch类方法。 - 用异常捕获+受影响行数双重校验:
- 用
try-except包裹cursor.execute()执行逻辑,所有SQL执行错误(语法错误、权限不足、表已存在等)都会以pyodbc.Error子类异常的形式抛出,进入异常分支即可判定执行失败。你之前的报错被fetch调用的异常掩盖,移除fetch代码后就能拿到SQL本身的具体错误信息。 - 执行无异常时,读取
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
相关产品推荐
相关产品推荐

