运行在Heroku上的Django项目如何上传文件至AWS S3

Heroku dyno的文件系统是临时、只读的，仅根目录下的/tmp目录支持临时写入，且dyno重启后该目录下所有文件会被自动清空，完全不适合存储持久化文件。
你最初的代码逻辑是先把CSV写入本地磁盘路径，再调用S3接口读取本地文件上传，本身就不符合Heroku的运行规则；加上代码里写的是相对路径tmp/{task_id}.csv，没有用/tmp的绝对路径，就算放开写入权限，运行目录下不存在对应tmp文件夹也会直接触发FileNotFoundError报错。

完全跳过本地磁盘写入步骤，在内存中生成CSV内容后直接调用S3接口上传，全程不碰本地文件系统，没有磁盘IO开销，完美适配Heroku这类只读文件系统的部署环境。
首先补全必要的依赖导入：

import csv
import os
import logging
from io import StringIO
from typing import List, Any
from celery import Celery
from faker import Faker
import boto3
from botocore.exceptions import ClientError

fake = Faker()
app = Celery("your_app_name")
logger = logging.getLogger(__name__)

修正后的完整任务代码如下：

@app.task
def upload_to_s3(file_name: str, bucket: str, file_buffer: StringIO) -> bool:
    """将内存中的文件内容直接上传到S3"""
    s3_client = boto3.client("s3")
    try:
        s3_client.put_object(
            Body=file_buffer.getvalue(),
            Bucket=bucket,
            Key=f"celery_upload/{file_name}.csv",
            ContentType="text/csv"
        )
    except ClientError as e:
        logger.error(f"S3上传失败: {str(e)}")
        return False
    return True

@app.task
def generate_csv_buffer(data: List[List[Any]]) -> StringIO:
    """在内存中生成CSV内容，不写入本地磁盘"""
    headers = ["name", "phone", "email"]
    buffer = StringIO()
    writer = csv.writer(
        buffer,
        delimiter=";",
        quotechar='"',
        quoting=csv.QUOTE_MINIMAL
    )
    writer.writerow(headers)
    writer.writerows(data)
    return buffer

@app.task(bind=True)
def generate_fake_data(self, total: int):
    """生成测试数据并上传S3的主流程"""
    # 生成假数据
    fake_data = []
    for _ in range(total):
        fake_data.append([
            fake.name(),
            fake.phone_number(),
            fake.email()
        ])
    # 内存中生成CSV
    csv_buffer = generate_csv_buffer(fake_data)
    # 直接上传，全程无本地文件操作
    upload_success = upload_to_s3(
        file_name=self.request.id,
        bucket=os.getenv("AWS_STORAGE_BUCKET_NAME"),
        file_buffer=csv_buffer
    )
    # 关闭内存缓冲区释放资源
    csv_buffer.close()
    if not upload_success:
        raise Exception("CSV上传S3失败")
    return f"已生成并上传{total}条测试数据，对应S3文件路径：celery_upload/{self.request.id}.csv"

• 上传S3时建议加上ContentType="text/csv"参数，否则后续访问文件时S3默认会返回application/octet-stream类型，可能导致浏览器下载或者打开文件乱码。
• 内存缓冲区用完后记得调用close()释放资源，避免Celery worker长时间运行出现内存泄漏。
• 确认运行Celery worker的环境配置了正确的S3访问权限，和你原有Django静态文件上传用的AWS密钥、权限保持一致即可。
• 如果需要处理GB级别的超大CSV文件，内存占用过高不适合用StringIO方案，可以临时将文件写入绝对路径/tmp/{task_id}.csv，上传完成后主动删除临时文件，避免占满临时目录空间，注意不要依赖这个目录做持久化存储。

内容的提问来源于stack exchange，提问作者Vitalii Mytenko