mirror of
https://github.com/wahyd4/links.git
synced 2026-08-08 21:04:53 +10:00
47 lines
1.3 KiB
Python
47 lines
1.3 KiB
Python
import boto3
|
|
from django.conf import settings
|
|
from botocore.config import Config
|
|
|
|
class R2Storage:
|
|
def __init__(self):
|
|
self.client = boto3.client('s3',
|
|
endpoint_url=settings.R2_ENDPOINT_URL,
|
|
aws_access_key_id=settings.R2_ACCESS_KEY_ID,
|
|
aws_secret_access_key=settings.R2_SECRET_ACCESS_KEY,
|
|
config=Config(signature_version='s3v4'),
|
|
region_name='auto'
|
|
)
|
|
self.bucket = settings.R2_BUCKET_NAME
|
|
|
|
def upload_file(self, file_obj, key, content_type=None):
|
|
"""Upload a file to R2"""
|
|
extra_args = {}
|
|
if content_type:
|
|
extra_args['ContentType'] = content_type
|
|
|
|
self.client.upload_fileobj(
|
|
file_obj,
|
|
self.bucket,
|
|
key,
|
|
ExtraArgs=extra_args
|
|
)
|
|
return key
|
|
|
|
def delete_file(self, key):
|
|
"""Delete a file from R2"""
|
|
self.client.delete_object(
|
|
Bucket=self.bucket,
|
|
Key=key
|
|
)
|
|
|
|
def get_url(self, key, expires_in=3600):
|
|
"""Generate a presigned URL for the file"""
|
|
return self.client.generate_presigned_url(
|
|
'get_object',
|
|
Params={
|
|
'Bucket': self.bucket,
|
|
'Key': key
|
|
},
|
|
ExpiresIn=expires_in
|
|
)
|