如何实现SQL查询依次执行且出错时仍继续后续查询？

把每个查询单独放入try-except块，或者用查询列表+循环的方式实现，这两种方案都完全符合你的需求，合理性没问题，后者更是更优的选择，下面具体分析：

1. 单个try-except包裹单查询的方案 #

这种方案逻辑直白，每个查询的错误可以单独处理（比如某条查询出错后要做特定的告警或数据记录），但缺点是代码冗余度高——每个查询都要写一遍try-except、execute、rowcount获取的逻辑，后续新增或修改查询时容易重复劳动。

2. 查询列表+循环的方案（推荐优先使用） #

这是单个try-except方案的优化版，完美解决冗余问题，实现起来也很简单：

• 核心思路：把所有需要执行的查询（可以附带查询名称、参数等信息）整理成一个列表，然后循环遍历列表，每次循环内用try-except处理单个查询的执行。
• 示例代码：

import pymysql

# 整理查询列表，附带名称方便排查问题，参数用元组传递
queries = [
    ("用户表统计", "SELECT COUNT(*) FROM users WHERE status = %s", (1,)),
    ("订单表查询", "SELECT id, order_no FROM orders WHERE create_time > %s", ("2024-01-01",)),
    # 新增查询直接往列表里加就行
]

# 用上下文管理器管理连接和游标，自动关闭资源
with pymysql.connect(host='localhost', user='your_user', password='your_pass', db='your_db') as db:
    with db.cursor() as cursor:
        for query_name, sql, params in queries:
            try:
                cursor.execute(sql, params)
                row_count = cursor.rowcount
                print(f"✅ {query_name} 执行成功，影响行数: {row_count}")
                # 如果需要查询结果，可添加 result = cursor.fetchall()
            except pymysql.MySQLError as e:
                print(f"❌ {query_name} 执行失败: {str(e)}")
                # 生产环境建议用logging模块记录错误，替代print

• 优势：
  • 代码简洁易维护，新增查询只需往列表追加条目，无需重复写异常处理逻辑
  • 统一的错误处理逻辑，也可根据query_name做差异化处理（比如某类查询出错后重试）
  • 结合上下文管理器with，能自动释放数据库连接和游标资源，避免资源泄漏

额外优化建议 #

• 精准捕获异常：尽量捕获pymysql.MySQLError而非宽泛的Exception，避免把非数据库相关的错误（比如变量名写错）当成数据库查询错误处理
• 封装执行函数：如果逻辑更复杂，可以把单条查询的执行逻辑封装成函数，比如：

def run_query(cursor, query_name, sql, params):
    try:
        cursor.execute(sql, params)
        return (True, cursor.rowcount)
    except pymysql.MySQLError as e:
        print(f"{query_name} 失败: {str(e)}")
        return (False, str(e))

然后循环里直接调用这个函数，代码结构更清晰

• 日志替代打印：生产环境用Python的logging模块记录执行日志，方便后续排查问题，比print更灵活规范

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