Python遇到数据库报错后如何循环重试运行代码直至错误消除？

基础实现方案 #

核心逻辑是捕获数据库专属异常，通过循环执行操作，异常触发时等待固定间隔后重试，直到操作成功。你可以根据需求选择无限重试，或者设置最大重试次数避免无意义等待：

import time
import pymysql
from pymysql.err import OperationalError  # 不同数据库驱动替换为对应连接异常类即可

# 自定义配置参数
MAX_RETRIES = None  # 设为None表示无限重试，设为正整数则为最大重试次数
RETRY_INTERVAL = 5  # 每次重试的等待间隔，单位秒

retry_count = 0
while True:
    try:
        # 替换为你自己的数据库连接+业务逻辑代码
        conn = pymysql.connect(
            host="你的数据库地址",
            user="数据库用户名",
            password="数据库密码",
            database="目标库名"
        )
        cursor = conn.cursor()
        cursor.execute("SELECT 1")  # 测试连通性
        # 此处写你的业务逻辑
        print("数据库操作执行成功")
        break  # 执行无异常直接跳出重试循环
    except OperationalError as e:
        retry_count += 1
        print(f"数据库连接异常，第{retry_count}次重试，错误信息：{e}")
        # 配置了最大重试次数时，达到上限直接抛出错误终止程序
        if MAX_RETRIES is not None and retry_count >= MAX_RETRIES:
            raise Exception(f"已达到最大重试次数{MAX_RETRIES}，数据库连接失败")
        time.sleep(RETRY_INTERVAL)

进阶：指数退避重试实现 #

如果想要更合理的重试策略，避免短时间内大量请求打爆数据库，可以用指数退避的间隔规则：初始等待时间短，每次重试间隔翻倍，达到阈值后封顶：

import time
import pymysql
from pymysql.err import OperationalError

MAX_RETRIES = 10
INITIAL_RETRY_DELAY = 1  # 首次重试等待1秒
MAX_RETRY_DELAY = 30  # 最大等待30秒

retry_count = 0
current_delay = INITIAL_RETRY_DELAY
while True:
    try:
        # 替换为你自己的数据库操作代码
        conn = pymysql.connect(host="xxx", user="xxx", password="xxx", database="xxx")
        # 业务逻辑执行
        break
    except OperationalError as e:
        retry_count += 1
        print(f"连接失败，第{retry_count}次重试，等待{current_delay}秒，错误：{e}")
        if retry_count >= MAX_RETRIES:
            raise Exception(f"重试{MAX_RETRIES}次后仍无法连接数据库")
        time.sleep(current_delay)
        current_delay = min(current_delay * 2, MAX_RETRY_DELAY)

注意事项 #

• 只捕获数据库连接、网络相关的特定异常，不要直接捕获所有Exception，避免把SQL语法错误、权限错误这类不需要重试的逻辑问题也纳入重试范围。
• 如果是长期运行的后台服务，除了启动阶段的连接重试，运行过程中连接断开的场景也可以复用这套逻辑做自动重连。

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