Python实现每N分钟自动轮换AWS临时凭证的代码改造问询

核心逻辑通过时间戳校验或后台定时线程，按照你设定的N分钟间隔自动重新获取AWS token生成最新的S3客户端，无需手动触发刷新，提供两种改造方案可根据你的业务场景选择：

方案1：主动校验刷新（推荐，兼容性最高） #

适用于可以调整bucket_parse_function内部逻辑的场景，每次调用S3操作前自动判断是否需要刷新凭证：

import time

# 定义凭证刷新间隔，单位分钟，替换为你需要的N值
REFRESH_INTERVAL_MINUTES = 5
_last_refresh_time = 0
_s3_client = None

def get_refreshed_s3_client():
    global _last_refresh_time, _s3_client
    now = time.time()
    # 计算距离上次刷新的时间差
    elapsed_minutes = (now - _last_refresh_time) / 60
    
    # 首次初始化或达到刷新间隔时重新获取凭证
    if not _s3_client or elapsed_minutes >= REFRESH_INTERVAL_MINUTES:
        try:
            aws_token = get_token(account_id=account_id, region="us-east-1")
            _s3_client = get_boto_client_with_creds('s3', aws_token=aws_token)
            _last_refresh_time = now
            print("已刷新AWS凭证", _s3_client)
        except Exception as e:
            print("获取AWS凭证失败：", e)
            raise # 抛出异常避免使用过期客户端执行业务
    return _s3_client

if valid_bucket(bucket_name):
    s3 = get_refreshed_s3_client()
    # 如果bucket_parse_function是循环任务，把get_refreshed_s3_client放到循环内每次操作前调用即可
    bucket_parse_function(s3)

方案2：后台线程自动刷新 #

适用于无法修改bucket_parse_function内部逻辑的场景，通过守护线程在后台自动按间隔刷新凭证：

import time
import threading

REFRESH_INTERVAL_MINUTES = 5
_s3_client = None
# 加锁保证线程安全，避免读写s3客户端时出现竞争问题
_refresh_lock = threading.Lock()

def refresh_worker():
    global _s3_client
    while True:
        try:
            aws_token = get_token(account_id=account_id, region="us-east-1")
            with _refresh_lock:
                _s3_client = get_boto_client_with_creds('s3', aws_token=aws_token)
            print("后台已刷新AWS凭证")
        except Exception as e:
            print("凭证刷新失败，等待重试：", e)
        time.sleep(REFRESH_INTERVAL_MINUTES * 60)

if valid_bucket(bucket_name):
    # 启动后台刷新线程，设置为守护线程，主程序退出时自动销毁
    refresh_thread = threading.Thread(target=refresh_worker, daemon=True)
    refresh_thread.start()
    # 等待首次凭证初始化完成
    while not _s3_client:
        time.sleep(0.1)
    bucket_parse_function(_s3_client)

• 可以把刷新间隔设置得比AWS临时凭证的实际过期时间短1-2分钟，提前刷新避免出现凭证过期报错
• 如果bucket_parse_function单次执行耗时超过N分钟，建议在函数内部所有S3操作前都调用get_refreshed_s3_client获取最新客户端
• 后台线程方案需要保证get_token接口调用不会阻塞过长时间，避免刷新延迟

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