Python脚本执行PostgreSQL权限授予遇并发更新错误的重试方案咨询

针对执行GRANT SELECT ON ALL TABLES IN SCHEMA...时偶尔出现的psycopg2.errors.InternalError_: tuple concurrently updated并发冲突问题，以下是基于延迟重试的解决方案，利用指数退避策略给数据库足够时间自行恢复：

核心实现思路 #

• 只针对特定的并发冲突异常重试，避免无差别处理其他错误
• 采用指数退避延迟（1s→2s→4s），减少对数据库的重复压力
• 限制最大重试次数，防止无限循环
• 每次重试重新建立数据库连接，避免使用处于异常状态的会话

代码实现 #

1. 重试装饰器定义 #

import psycopg2
from psycopg2 import errors
import time

def retry_on_tuple_conflict(max_retries=3, initial_delay=1):
    def decorator(func):
        def wrapper(*args, **kwargs):
            retries = 0
            delay = initial_delay
            while retries < max_retries:
                try:
                    return func(*args, **kwargs)
                except errors.InternalError_ as e:
                    # 仅捕获目标并发冲突异常
                    if "tuple concurrently updated" in str(e):
                        retries += 1
                        if retries >= max_retries:
                            raise  # 达到重试上限，抛出原异常
                        time.sleep(delay)
                        delay *= 2  # 指数退避延迟翻倍
                    else:
                        raise  # 其他内部错误直接抛出
                except Exception as e:
                    raise  # 非目标异常不重试
        return wrapper
    return decorator

2. 授权函数（带重试） #

@retry_on_tuple_conflict(max_retries=3, initial_delay=1)
def grant_select_to_role(conn_params, role_name, schema_name="public"):
    # 每次重试重新创建连接，确保会话干净
    with psycopg2.connect(**conn_params) as conn:
        with conn.cursor() as cur:
            # 执行现有表的权限授予
            grant_sql = f"GRANT SELECT ON ALL TABLES IN SCHEMA {schema_name} TO {role_name};"
            cur.execute(grant_sql)
            # 可选：授予未来创建表的默认权限
            alter_default_sql = f"ALTER DEFAULT PRIVILEGES IN SCHEMA {schema_name} GRANT SELECT ON TABLES TO {role_name};"
            cur.execute(alter_default_sql)
            conn.commit()

3. 使用示例 #

# 数据库连接参数
conn_params = {
    "dbname": "your_database",
    "user": "your_admin_user",
    "password": "your_password",
    "host": "localhost",
    "port": 5432
}

# 调用授权函数
try:
    grant_select_to_role(conn_params, "new_db_role")
    print("权限授予成功")
except Exception as e:
    print(f"最终失败: {str(e)}")

注意事项 #

• 若需要授予视图、序列等对象的权限，可修改GRANT语句为：GRANT SELECT ON ALL TABLES, SEQUENCES, VIEWS IN SCHEMA {schema_name} TO {role_name};
• 重试次数和初始延迟可根据实际业务场景调整，比如高并发环境可适当增加延迟
• 确保连接参数中的用户拥有足够的权限执行GRANT操作

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