Python Celery任务中Boto3获取SecretsManager密钥失败问题排查

核心问题分析及解决步骤 #

• Boto3客户端初始化位置错误
  你当前在Celery任务外部初始化client_secrets，但Celery Worker进程与Web主进程是完全隔离的，主进程创建的客户端无法在Worker进程中正常复用，大概率会导致网络连接阻塞，表现为任务无限加载。
  解决： 将客户端初始化逻辑移到Celery任务内部，每次任务执行时创建新客户端：

  @shared_task(name="get_token")
  def get_token(authCode):
      try:
          # 任务内部初始化Boto3客户端
          client_secrets = boto3.client('secretsmanager', region_name="us-east-1")
          get_secret_value_response = client_secrets.get_secret_value(
              SecretId=secret_name
          )
          database_secrets = json.loads(get_secret_value_response['SecretString'])
          print(database_secrets['password'])
      except ClientError as e:
          raise e
  

• Celery Worker未加载AWS凭据
  虽然你通过aws configure配置了本地凭据，但Celery Worker启动时的环境可能未继承这些配置，导致无法完成AWS身份验证，任务因阻塞在权限校验环节而挂起。
  解决：

  • 启动Worker时显式注入AWS环境变量：

    AWS_ACCESS_KEY_ID=你的密钥ID AWS_SECRET_ACCESS_KEY=你的密钥 celery -A 你的应用名 worker --loglevel=info
    

  • 或在代码中通过环境变量读取凭据（推荐，避免硬编码）：

    import os
    client_secrets = boto3.client(
        'secretsmanager',
        region_name="us-east-1",
        aws_access_key_id=os.getenv('AWS_ACCESS_KEY_ID'),
        aws_secret_access_key=os.getenv('AWS_SECRET_ACCESS_KEY')
    )
    

• 代码中的语法/逻辑错误
  你的代码存在两处明显错误，会导致任务执行失败：

  1. 任务定义get_token()未接收authCode参数，但视图调用时传递了该参数，会触发参数不匹配的错误；
  2. 解析密钥时使用了未定义的变量response，实际应该用get_secret_value_response。

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