Python数据入AWS S3：密钥管理与Token复用问题咨询

1. 密钥管理方案选择 #

本地配置文件（适合开发/测试环境） #

优点是上手简单，适合本地调试，但绝对不能把配置文件提交到代码仓库，必须加到.gitignore里避免泄露。
示例代码（用YAML格式）：

import yaml

# 读取本地配置
with open('config.yaml', 'r') as f:
    config = yaml.safe_load(f)

# 提取密钥
saas_username = config['saas']['username']
saas_password = config['saas']['password']
saas_api_key = config['saas']['api_key']

config.yaml示例内容：

saas:
  username: "your_username"
  password: "your_password"
  api_key: "your_api_key"
aws:
  s3_bucket: "your-bucket-name"

AWS Secrets Manager（适合生产环境） #

生产环境优先用这个，密钥存储在云端，权限可精细化控制，彻底避免本地文件泄露风险。需要给运行代码的角色（比如Lambda执行角色、ECS任务角色）配置secretsmanager:GetSecretValue权限。
示例代码：

import boto3
import json

def get_saas_secrets():
    secret_name = "saas-api-credentials"
    region_name = "us-east-1"  # 替换为你的AWS区域

    session = boto3.session.Session()
    client = session.client(service_name='secretsmanager', region_name=region_name)

    secret_response = client.get_secret_value(SecretId=secret_name)
    return json.loads(secret_response['SecretString'])

# 获取密钥
saas_creds = get_saas_secrets()
saas_username = saas_creds['username']
saas_password = saas_creds['password']
saas_api_key = saas_creds['api_key']

2. Token管理优化（100个Endpoint复用逻辑） #

核心思路是把Token的获取、缓存、刷新逻辑封装成通用工具，所有Endpoint共用这套逻辑，避免重复写100次流水线。

通用API客户端示例代码 #

import requests
import time
import json
import boto3

class SaaSAPIClient:
    def __init__(self, username, password, api_key):
        self.username = username
        self.password = password
        self.api_key = api_key
        self.token = None
        self.token_expiry = 0  # 记录Token过期时间戳

    def _fetch_new_token(self):
        # 调用SaaS的Token获取接口
        auth_url = "https://saas-server.com/auth/token"
        payload = {
            "username": self.username,
            "password": self.password,
            "api_key": self.api_key
        }
        response = requests.post(auth_url, json=payload)
        response.raise_for_status()  # 请求失败时抛出异常
        token_data = response.json()
        self.token = token_data['access_token']
        # 提前100秒刷新Token，避免刚好过期时请求失败
        self.token_expiry = time.time() + token_data['expires_in'] - 100

    def _ensure_valid_token(self):
        # 检查Token是否过期，过期则重新获取
        if not self.token or time.time() >= self.token_expiry:
            self._fetch_new_token()

    def call_endpoint(self, endpoint_url, method="GET", payload=None):
        # 确保Token有效
        self._ensure_valid_token()
        # 构造请求头
        headers = {
            "Authorization": f"Bearer {self.token}",
            "X-API-Key": self.api_key
        }
        # 发送请求
        response = requests.request(method, endpoint_url, json=payload, headers=headers)
        response.raise_for_status()
        return response.json()

# 初始化客户端
client = SaaSAPIClient(
    username=saas_username,
    password=saas_password,
    api_key=saas_api_key
)

# 初始化S3客户端
s3 = boto3.client('s3')

批量处理100个Endpoint #

把所有Endpoint的信息整理成列表，循环调用即可，不需要重复写Token逻辑：

# 整理所有需要同步的Endpoint
endpoints = [
    {"url": "https://saas-server.com/api/users", "s3_key": "users/latest_data.json"},
    {"url": "https://saas-server.com/api/orders", "s3_key": "orders/latest_data.json"},
    {"url": "https://saas-server.com/api/products", "s3_key": "products/latest_data.json"},
    # ... 剩下97个Endpoint依次添加
]

# 批量同步数据到S3
for item in endpoints:
    try:
        # 调用API获取数据
        data = client.call_endpoint(item['url'])
        # 把数据存入S3
        s3.put_object(
            Bucket="your-bucket-name",
            Key=item['s3_key'],
            Body=json.dumps(data, indent=2)
        )
        print(f"✅ 成功同步 {item['url']} 到S3")
    except Exception as e:
        print(f"❌ 同步 {item['url']} 失败: {str(e)}")

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