s3 uploader

etl/s3/s3_uploader.py

@@ -0,0 +1,70 @@
+import os
+import boto3
+from botocore.exceptions import ClientError
+from urllib.parse import urlparse
+from datetime import datetime
+
+
+class S3Uploader:
+    """
+    Simple helper to upload local files to S3 and return their S3 HTTPS URI.
+    """
+
+    def __init__(
+        self,
+        aws_access_key: str = "AKIAU5A36PPNK7RXX52V",
+        aws_secret_key: str = "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",
+        region: str = "eu-west-2",
+    ):
+        self.aws_access_key = aws_access_key
+        self.aws_secret_key = aws_secret_key
+        self.region = region
+
+        self.s3 = boto3.client(
+            "s3",
+            aws_access_key_id=self.aws_access_key,
+            aws_secret_access_key=self.aws_secret_key,
+            region_name=self.region,
+        )
+
+    def upload_file(self, file_path: str, bucket: str, prefix: str = "uploads/") -> str:
+        """
+        Upload a local file to an S3 bucket and return its HTTPS URI.
+
+        Args:
+            file_path (str): Path to the local file.
+            bucket (str): S3 bucket name.
+            prefix (str): Folder/prefix in the bucket.
+
+        Returns:
+            str: HTTPS-style S3 URI (not signed).
+        """
+        try:
+            filename = os.path.basename(file_path)
+            timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
+            s3_key = os.path.join(prefix, f"{timestamp}_{filename}")
+
+            self.s3.upload_file(file_path, bucket, s3_key)
+
+            s3_uri = f"https://{bucket}.s3.{self.region}.amazonaws.com/{s3_key}"
+            return s3_uri
+
+        except ClientError as e:
+            raise RuntimeError(f"❌ S3 upload failed: {e}")
+        
+    def print_bucket(self):
+        print(self.s3.head_bucket(Bucket="retrofit-data-dev"))
+
+
+    def generate_presigned_url(self, bucket: str, key: str, expires_in: int = 3600) -> str:
+        """
+        Generate a temporary presigned URL for an S3 object.
+        """
+        try:
+            return self.s3.generate_presigned_url(
+                "get_object",
+                Params={"Bucket": bucket, "Key": key},
+                ExpiresIn=expires_in,
+            )
+        except ClientError as e:
+            raise RuntimeError(f"❌ Failed to generate signed URL: {e}")