如何根据运行环境动态获取对应AWS S3存储桶名称？Python实现

针对你的需求（脚本要在DEV/TEST/PROD环境切换，部署在Lambda或Data Pipeline），以下是几种实用的动态配置方案：

一、使用环境变量（最推荐，适配所有部署环境） #

环境变量是AWS Lambda和Data Pipeline都支持的配置方式，无需修改代码逻辑，只需在部署时设置对应环境的变量值即可。

修改后的代码： #

import pandas as pd
import boto3
import os
from botocore.exceptions import ClientError

# 从环境变量获取存储桶名，默认值设为dev环境的桶（本地调试用）
AWS_S3_BUCKET = os.getenv("AWS_S3_BUCKET", "bucket-dev00")

def connect_s3(profile):
    """连接AWS S3"""
    session = boto3.Session(profile_name=profile)
    s3_client = session.client('s3', region_name='us-east-1')
    logger.info(f'S3 session successfully created')
    return s3_client

def get_file_from_s3(profile):
    """从S3获取文件并处理错误"""
    s3_client = connect_s3(profile)
    try:
        response = s3_client.get_object(Bucket=AWS_S3_BUCKET, Key="folder1/file.csv")
    except ClientError as e:
        raise e
    else:
        status = response.get("ResponseMetadata", {}).get("HTTPStatusCode")
        if status == 200:
            df = pd.read_csv(response.get("Body"), sep=',', quotechar='"', skipinitialspace=True)
            logger.info(f'Successful S3 get_object response. Status - {status}')
            return df
        else:
            logger.info(f'Unsuccessful S3 get_object response. Status - {status}')

部署配置： #

• AWS Lambda：在Lambda控制台的「配置」→「环境变量」中添加AWS_S3_BUCKET，对应环境设为bucket-dev00/bucket-tst00/bucket-prod00；如果用SAM/CloudFormation部署，直接在模板的环境变量字段指定对应值。
• AWS Data Pipeline：在创建活动时，通过「参数」或「环境变量」配置项传入AWS_S3_BUCKET的值。

二、基于传入参数动态映射 #

如果需要在调用函数时手动指定环境，可以通过函数参数传递环境标识，再映射到对应的存储桶名。

修改后的代码： #

import pandas as pd
import boto3
from botocore.exceptions import ClientError

# 定义环境与存储桶的映射关系
ENV_BUCKET_MAP = {
    "dev": "bucket-dev00",
    "test": "bucket-tst00",
    "prod": "bucket-prod00"
}

def connect_s3(profile):
    """连接AWS S3"""
    session = boto3.Session(profile_name=profile)
    s3_client = session.client('s3', region_name='us-east-1')
    logger.info(f'S3 session successfully created')
    return s3_client

def get_file_from_s3(profile, environment="dev"):
    """从S3获取文件并处理错误，支持指定环境"""
    # 校验环境参数合法性
    if environment not in ENV_BUCKET_MAP:
        raise ValueError(f"Invalid environment: {environment}. Allowed values: {list(ENV_BUCKET_MAP.keys())}")
    
    bucket_name = ENV_BUCKET_MAP[environment]
    s3_client = connect_s3(profile)
    try:
        response = s3_client.get_object(Bucket=bucket_name, Key="folder1/file.csv")
    except ClientError as e:
        raise e
    else:
        status = response.get("ResponseMetadata", {}).get("HTTPStatusCode")
        if status == 200:
            df = pd.read_csv(response.get("Body"), sep=',', quotechar='"', skipinitialspace=True)
            logger.info(f'Successful S3 get_object response. Status - {status}')
            return df
        else:
            logger.info(f'Unsuccessful S3 get_object response. Status - {status}')

使用方式： #

调用函数时传入环境参数，比如get_file_from_s3("my-profile", "prod")即可使用生产环境的存储桶。

三、使用AWS Systems Manager Parameter Store（企业级集中管理） #

如果需要集中管理所有环境的配置（无需在每个部署环境单独设置），可以用SSM Parameter Store存储不同环境的存储桶名，代码运行时动态拉取。

步骤： #

1. 在AWS控制台的Systems Manager → Parameter Store中创建参数：
   • /dev/s3/bucket：值为bucket-dev00
   • /test/s3/bucket：值为bucket-tst00
   • /prod/s3/bucket：值为bucket-prod00
2. 给Lambda/Data Pipeline的执行角色添加ssm:GetParameter权限。

修改后的代码： #

import pandas as pd
import boto3
from botocore.exceptions import ClientError

def get_bucket_name_from_ssm(environment="dev"):
    """从SSM Parameter Store获取对应环境的存储桶名"""
    ssm_client = boto3.client('ssm', region_name='us-east-1')
    parameter_path = f"/{environment}/s3/bucket"
    try:
        response = ssm_client.get_parameter(Name=parameter_path, WithDecryption=False)
        return response['Parameter']['Value']
    except ClientError as e:
        raise e

def connect_s3(profile):
    """连接AWS S3"""
    session = boto3.Session(profile_name=profile)
    s3_client = session.client('s3', region_name='us-east-1')
    logger.info(f'S3 session successfully created')
    return s3_client

def get_file_from_s3(profile, environment="dev"):
    """从S3获取文件并处理错误"""
    bucket_name = get_bucket_name_from_ssm(environment)
    s3_client = connect_s3(profile)
    try:
        response = s3_client.get_object(Bucket=bucket_name, Key="folder1/file.csv")
    except ClientError as e:
        raise e
    else:
        status = response.get("ResponseMetadata", {}).get("HTTPStatusCode")
        if status == 200:
            df = pd.read_csv(response.get("Body"), sep=',', quotechar='"', skipinitialspace=True)
            logger.info(f'Successful S3 get_object response. Status - {status}')
            return df
        else:
            logger.info(f'Unsuccessful S3 get_object response. Status - {status}')

优势： #

• 集中管理所有环境的配置，修改时无需更新代码或重新部署
• 支持版本控制和权限管控，适合企业级场景

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