Spaces:
Sleeping
Sleeping
File size: 2,091 Bytes
479eb2f |
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 |
import io
import os
import boto3
from botocore.exceptions import NoCredentialsError
ACCESS_KEY = os.environ.get('ACCESS_KEY')
SECRET_KEY = os.environ.get('SECRET_KEY')
ENDPOINT = os.environ.get("ENDPOINT")
BUCKET_NAME = 'host'
def create_s3_client(access_key, secret_key, endpoint_url):
try:
s3_client = boto3.client(
's3',
aws_access_key_id=access_key,
aws_secret_access_key=secret_key,
endpoint_url=endpoint_url
)
return s3_client
except NoCredentialsError:
print("Credentials not available")
return None
# 上传文件到 S3（或兼容 S3 的服务，如 IDrive e2）
def upload_file_stream_to_s3(file_stream, bucket_name, key):
s3_client = create_s3_client(ACCESS_KEY, SECRET_KEY, ENDPOINT)
file_stream.seek(0)
s3_client.upload_fileobj(
file_stream,
bucket_name,
key
)
# with open("output/output.pdf", 'wb') as f:
# file_stream.seek(0)
# f.write(file_stream.getvalue())
print(f"{key} has been uploaded to {bucket_name}")
return True
def download_file_stream_from_s3(bucket_name, key):
fileobj = io.BytesIO()
s3_client = create_s3_client(ACCESS_KEY, SECRET_KEY, ENDPOINT)
s3_client.download_fileobj(
bucket_name,
key,
fileobj,
)
return fileobj
# FILE_PATH = "src/assets/InvitationTemplate.pdf"
# OBJECT_NAME = 'test.pdf'
# ACCESS_KEY = os.environ.get('ACCESS_KEY')
# SECRET_KEY = os.environ.get('SECRET_KEY')
# ENDPOINT = os.environ.get("ENDPOINT")
# BUCKET_NAME = 'host'
# FILE_PATH = "src/assets/InvitationTemplate.pdf"
# OBJECT_NAME = 'test.pdf'
# # 创建 S3 客户端
# s3_client = create_s3_client(ACCESS_KEY, SECRET_KEY, ENDPOINT)
# # 上传文件
# if s3_client:
# # upload_file_to_s3(s3_client, FILE_PATH, BUCKET_NAME, OBJECT_NAME)
# fileobj = download_file_stream_from_s3(s3_client, BUCKET_NAME, OBJECT_NAME)
# with open(OBJECT_NAME, 'wb') as f:
# f.write(fileobj.getvalue()) |