如何使用Python将URL获取的CSV文件直接上传到S3存储桶？

前提准备 #

你需要提前安装两个依赖库，同时配置好AWS访问凭证（可通过本地AWS配置文件、环境变量或运行环境的IAM角色自动获取，无需硬编码在代码中）：

pip install requests boto3

核心实现逻辑 #

通过requests的流式请求直接获取CSV文件流，不落地到本地磁盘，直接用boto3的文件对象上传接口把流传到S3即可：

import requests
import boto3

# 替换为你自己的参数
CSV_URL = "https://example.com/your/file.csv"
S3_BUCKET_NAME = "your-bucket-name"
S3_OBJECT_KEY = "path/to/save/in/s3.csv"

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

# 流式获取CSV内容，不下载到本地
with requests.get(CSV_URL, stream=True) as response:
    # 确保请求成功
    response.raise_for_status()
    # 处理gzip等压缩编码的内容，避免上传乱码
    response.raw.decode_content = True
    # 直接上传文件流到S3
    s3.upload_fileobj(response.raw, S3_BUCKET_NAME, S3_OBJECT_KEY)

常见优化说明 #

• 大文件适配：流式请求不会一次性加载整个文件到内存，也不会写入本地磁盘，完全满足无本地存储的自动化流程要求，支持GB级大文件传输。
• 异常处理补充：可以根据业务需求添加URL请求异常、S3上传异常的捕获逻辑，示例如下：

import requests
import boto3
from botocore.exceptions import ClientError

CSV_URL = "https://example.com/your/file.csv"
S3_BUCKET_NAME = "your-bucket-name"
S3_OBJECT_KEY = "path/to/save/in/s3.csv"

s3 = boto3.client("s3")

try:
    with requests.get(CSV_URL, stream=True, timeout=30) as response:
        response.raise_for_status()
        response.raw.decode_content = True
        s3.upload_fileobj(response.raw, S3_BUCKET_NAME, S3_OBJECT_KEY)
    print("上传完成")
except requests.exceptions.RequestException as e:
    print(f"获取CSV文件失败：{e}")
except ClientError as e:
    print(f"S3上传失败：{e}")

• 如需内存中处理CSV：如果需要先修改CSV内容再上传，可以直接读取流内容到io.BytesIO或csv模块处理，全程也不会涉及本地磁盘写入，示例：

import requests
import boto3
import csv
from io import StringIO, BytesIO

CSV_URL = "https://example.com/your/file.csv"
S3_BUCKET_NAME = "your-bucket-name"
S3_OBJECT_KEY = "path/to/save/in/s3.csv"

s3 = boto3.client("s3")

with requests.get(CSV_URL, stream=True) as response:
    response.raise_for_status()
    # 读取CSV内容到内存中处理
    lines = (line.decode("utf-8") for line in response.iter_lines())
    reader = csv.reader(lines)
    output = StringIO()
    writer = csv.writer(output)
    
    # 示例：跳过第一行表头再上传
    next(reader)
    for row in reader:
        writer.writerow(row)
    
    # 转成字节流上传
    output_bytes = BytesIO(output.getvalue().encode("utf-8"))
    s3.upload_fileobj(output_bytes, S3_BUCKET_NAME, S3_OBJECT_KEY)

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