File size: 3,609 Bytes
792ad00 | 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 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 | import boto3
from botocore.exceptions import ClientError
from core.config import settings
import logging
logger = logging.getLogger(__name__)
class S3Service:
def __init__(self):
self.s3_client = boto3.client(
's3',
aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
region_name=settings.AWS_REGION
)
self.bucket_name = settings.AWS_S3_BUCKET
def get_public_url(self, key: str):
"""
Generates the standard S3 public URL for a given key.
"""
return f"https://{self.bucket_name}.s3.{settings.AWS_REGION}.amazonaws.com/{key}"
def get_presigned_url(self, key: str, expires_in: int = 3600):
"""
Generates a pre-signed URL for secure access. Default: 1 hour.
"""
try:
url = self.s3_client.generate_presigned_url(
'get_object',
Params={'Bucket': self.bucket_name, 'Key': key},
ExpiresIn=expires_in
)
return url
except ClientError as e:
logger.error(f"Failed to generate presigned URL: {e}")
return None
async def upload_file(self, file_content: bytes, filename: str, user_id: str):
"""
Uploads a file to S3 under a user-specific folder.
"""
key = f"users/{user_id}/sources/{filename}"
try:
self.s3_client.put_object(
Bucket=self.bucket_name,
Key=key,
Body=file_content
)
return {
"key": key,
"public_url": self.get_public_url(key),
"private_url": self.get_presigned_url(key)
}
except ClientError as e:
logger.error(f"Failed to upload to S3: {e}")
raise Exception("S3 Upload Failed")
async def list_user_files(self, user_id: str):
"""
Lists files for a specific user.
"""
prefix = f"users/{user_id}/sources/"
try:
response = self.s3_client.list_objects_v2(
Bucket=self.bucket_name,
Prefix=prefix
)
files = []
if 'Contents' in response:
for obj in response['Contents']:
# Remove the prefix from the filename for display
filename = obj['Key'].replace(prefix, "")
if filename: # Avoid empty strings if the prefix itself is returned
files.append({
"filename": filename,
"key": obj['Key'],
"public_url": self.get_public_url(obj['Key']),
"private_url": self.get_presigned_url(obj['Key']),
"size": obj['Size'],
"last_modified": obj['LastModified']
})
return files
except ClientError as e:
logger.error(f"Failed to list S3 files: {e}")
raise Exception("S3 List Failed")
async def delete_file(self, key: str):
"""
Deletes a file from S3.
"""
try:
self.s3_client.delete_object(
Bucket=self.bucket_name,
Key=key
)
logger.info(f"Deleted S3 object: {key}")
return True
except ClientError as e:
logger.error(f"Failed to delete S3 object: {e}")
return False
s3_service = S3Service()
|