File size: 2,053 Bytes
50925ca | 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 | import boto3
from typing import Optional
from botocore.config import Config
from botocore.exceptions import ClientError
from app.config import settings
from loguru import logger
# Initialize boto3 S3 client for Supabase Storage (S3-compatible)
s3_client = boto3.client(
"s3",
endpoint_url=settings.CLOUDFLARE_R2_ENDPOINT_URL,
aws_access_key_id=settings.CLOUDFLARE_R2_ACCESS_KEY,
aws_secret_access_key=settings.CLOUDFLARE_R2_SECRET_KEY,
config=Config(signature_version="s3v4"),
region_name=settings.CLOUDFLARE_R2_REGION_NAME
)
def upload_file(file_content: bytes, object_key: str, content_type: str = "application/octet-stream") -> bool:
try:
s3_client.put_object(
Bucket=settings.CLOUDFLARE_R2_BUCKET,
Key=object_key,
Body=file_content,
ContentType=content_type
)
logger.info(f"Successfully uploaded {object_key} to {settings.CLOUDFLARE_R2_BUCKET}")
return True
except ClientError as e:
logger.error(f"Failed to upload {object_key} to S3: {e}")
return False
def get_file(object_key: str) -> Optional[bytes]:
try:
response = s3_client.get_object(
Bucket=settings.CLOUDFLARE_R2_BUCKET,
Key=object_key
)
return response["Body"].read()
except ClientError as e:
logger.error(f"Failed to download {object_key} from S3: {e}")
return None
def generate_download_url(object_key: str, expires_in: int = 3600) -> str:
try:
url = s3_client.generate_presigned_url(
"get_object",
Params={
"Bucket": settings.CLOUDFLARE_R2_BUCKET,
"Key": object_key
},
ExpiresIn=expires_in
)
return url
except ClientError as e:
logger.error(f"Failed to generate presigned URL for {object_key}: {e}")
# Fallback to direct construction if presigned fails
return f"{settings.CLOUDFLARE_R2_ENDPOINT_URL}/{settings.CLOUDFLARE_R2_BUCKET}/{object_key}"
|