如何在Django中实现数据库失败操作的自动重试？

问题背景 #

数据库操作偶尔会因临时网络故障或数据库短暂不可用失败，谷歌托管数据库文档要求：连接池需在连接异常时自动重连，应用程序需在事务中途出错时重试事务。已了解Django的连接管理机制，但不清楚如何在Django中实现事务重试逻辑。

当前遇到的具体问题：视图执行部分数据库操作后，后续操作随机触发连接错误，错误信息如下：

OperationalError at
consuming input failed: server closed the connection unexpectedly
This probably means the server terminated abnormally
before or while processing the request.

/.../django/forms/widgets.py, line 657, in optgroups
657. for index, (option_value, option_label) in enumerate(self.choices):
/.../django/forms/models.py, line 1403, in iter
1403. for obj in queryset:
/.../django/db/backends/utils.py, line 80, in _execute_with_wrappers
80. return executor(sql, params, many, context)
/.../django/db/backends/utils.py, line 84, in _execute
84. with self.db.wrap_database_errors:
/.../django/db/utils.py, line 91, in exit
91. raise dj_exc_value.with_traceback(traceback) from exc_value
/.../django/db/backends/utils.py, line 89, in _execute
89. return self.cursor.execute(sql, params)
/.../psycopg/server_cursor.py, line 294, in execute
294. raise ex.with_traceback(None)

该错误仅在约0.001%的查询中随机出现，无需排查根本原因，只需通过重试操作让视图最终成功渲染。

实现方案 #

1. 自定义事务重试装饰器 #

针对视图或事务函数，实现带指数退避的重试逻辑，只捕获明确的临时数据库错误：

import time
from functools import wraps
from django.db import transaction, OperationalError

def retry_transaction(max_retries=3, initial_delay=0.1):
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            retries = 0
            while retries < max_retries:
                try:
                    # 原子事务确保操作全成功或全回滚
                    with transaction.atomic():
                        return func(*args, **kwargs)
                except OperationalError as e:
                    # 只处理连接断开类临时错误
                    if "server closed the connection unexpectedly" in str(e):
                        retries += 1
                        if retries == max_retries:
                            raise
                        # 指数退避延迟，避免频繁重试压垮数据库
                        delay = initial_delay * (2 ** retries)
                        time.sleep(delay)
                    else:
                        # 非临时错误直接抛出，不重试
                        raise
        return wrapper
    return decorator

2. 在视图中应用装饰器 #

将装饰器直接套用到需要重试的视图或数据库操作函数上：

from django.shortcuts import render

@retry_transaction(max_retries=3)
def my_view(request):
    # 这里放置你的数据库操作逻辑
    # 比如查询集遍历、表单数据写入等
    return render(request, 'template.html', context)

3. 注意事项 #

• 仅对幂等操作使用重试：比如查询、更新已存在数据这类操作，避免重试创建订单这类非幂等操作导致重复数据。
• 控制重试阈值：不要设置过多重试次数，指数退避的延迟时间也要合理，避免给数据库额外施压。
• 精准匹配错误：只捕获明确的临时连接错误，不要对所有OperationalError重试（比如权限不足、语法错误这类问题重试也没用）。

内容的提问来源于stack exchange，提问作者Koen De Wit