Как загрузить файл на AWS S3 с помощью Django, работающего на Heroku?
У меня есть приложение, развернутое на Heroku.
Я следовал руководству по ссылке для настройки загрузки статических файлов на S3, и это работает. Теперь мне нужно загрузить CSV-файл, который был создан задачей celery, и загрузить его на S3. Проблема в том, что файловая система Heroku доступна только для чтения, и я не могу сохранить файл на ней. Следовательно, я получаю ошибку FileNotFoundError: [Errno 2] No such file or directory: 'tmp/a30113c5-bbbc-4432-9826-3918e547d407.csv'
.
Как мне быть?
@app.task
def upload_file(file_name, bucket, object_name=None):
"""Upload a file to an S3 bucket."""
# If S3 object_name was not specified, use file_name
if object_name is None:
object_name = os.path.basename(file_name)
# Upload the file
s3_client = boto3.client("s3")
try:
response = s3_client.upload_file(file_name, bucket, object_name)
except ClientError as e:
logging.error(e)
return False
return True
@app.task
def make_csv(data: List[Any], task_id: str):
"""Produce csv file with generated fake data and name it as task id."""
headers: List[str] = ["name", "phone", "email"]
file_path = os.path.normpath(f"tmp/{task_id}.csv")
with open(file=file_path, mode="w", encoding="UTF-8", newline="") as csv_file:
writer = csv.writer(
csv_file, delimiter=";", quotechar='"', quoting=csv.QUOTE_MINIMAL
)
writer.writerow(headers)
writer.writerows(data)
return csv_file
@app.task(bind=True)
def generate_fake_data(self, total: int):
"""Generate fake data function."""
fake_data: List[Any] = []
for _ in range(total):
name = fake.name()
phone = fake.phone_number()
email = fake.email()
fake_data.append([name, phone, email])
csv_file = make_csv(data=fake_data, task_id=self.request.id)
upload_file(
file_name=csv_file,
bucket=os.getenv("AWS_STORAGE_BUCKET_NAME"),
)
return f"{total} random data rows created."
Если файловая система доступна только для чтения, то сохраните файл в памяти (используя StringIO
) и загрузите его на s3 соответственно:
from io import StringIO
@app.task
def make_csv(data: List[Any], task_id: str):
"""Produce csv file with generated fake data and name it as task id."""
file_path = os.path.normpath(f"tmp/{task_id}.csv")
csv_file = StringIO()
writer = csv.writer(
csv_file, delimiter=";", quotechar='"', quoting=csv.QUOTE_MINIMAL
)
writer.writerow(headers)
writer.writerows(data)
return csv_file
В вашей задаче upload_file
используйте метод put_object
вместо upload_file
, и отправьте содержимое в виде байтов:
@app.task
def upload_file(file, bucket, object_name=None):
...
s3_client = boto3.client("s3")
raw_file = bytes(file.getvalue()) # convert to bytes
try:
response = s3_client.put_object(raw_file, bucket, object_name)
except ClientError as e:
logging.error(e)
return False
return True