File size: 1,805 Bytes
654965f
 
1979816
654965f
 
 
 
fed5666
 
 
654965f
 
1979816
 
654965f
 
 
fed5666
 
 
654965f
1979816
 
654965f
 
fed5666
654965f
 
 
1979816
654965f
 
 
d8b5430
654965f
 
 
 
fed5666
654965f
 
 
 
 
 
 
 
1979816
 
 
 
 
654965f
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
import base64
import io
import os
import boto3
from botocore.config import Config
import random
import string
from config_settings import settings




class S3ManagerService:
    def __init__(self):
        self.s3 = boto3.client(
            "s3",
            config=Config(signature_version="s3v4"),
            aws_access_key_id=settings.AWS_ACCESS_KEY_ID,
            aws_secret_access_key=settings.AWS_SECRET_ACCESS_KEY,
            region_name=settings.AWS_REGION,
        )
    
    def generate_signed_url(self, file_name: str, exp: int = 43200) -> str:  # 43200 seconds = 12 hours
        return self.s3.generate_presigned_url(
            "get_object",
            Params={"Bucket": settings.AWS_BUCKET_NAME, "Key": file_name},
            ExpiresIn=exp,
        )

    def generate_unique_file_name(self, file_name: str) -> str:
        random_string = "".join(
            random.choices(string.ascii_uppercase + string.digits, k=10)
        )
        file_extension = "png"
        file_real_name = file_name.split(".")[0]
        return f"{file_real_name}-{random_string}.{file_extension}"

    def upload_file(self, file, file_name) -> str:
        self.s3.upload_fileobj(file, settings.AWS_BUCKET_NAME, file_name)
        return file_name

    def upload_base64_file(self, base64_file: str, file_name: str) -> str:
        return self.upload_file(io.BytesIO(base64.b64decode(base64_file)), file_name)

    def get_object(self, file_name: str, bucket: str):
        try:
            return self.s3.get_object(Bucket=bucket, Key=file_name)
        except self.s3.exceptions.NoSuchKey:
            print(f"The file {file_name} does not exist in the bucket {bucket}.")
            return None
        except Exception as e:
            print(f"An error occurred: {e}")
            return None