Добавил новый подход к загрузке файла через S3

app/core/uploader.py

@@ -1,5 +1,4 @@
-from app.utils.uploader import S3ChunkUploadBackend # Используется для подключения S3
+from app.utils.uploader import HybridDiskS3UploadBackend
-# from app.utils.uploader import DiskChunkUploadBackend
 
 
-uploader_backend = S3ChunkUploadBackend()
+uploader_backend = HybridDiskS3UploadBackend()

app/utils/uploader.py

@@ -119,3 +119,41 @@ class S3ChunkUploadBackend(ChunkUploadBackend):
             )
         self.part_number = 1
         self.parts = []
+
+
+class HybridDiskS3UploadBackend(ChunkUploadBackend):
+    def __init__(self, key_prefix: str = ""):
+        self.s3 = boto3.client(
+            service_name="s3",
+            aws_access_key_id=settings.S3_ACCESS_KEY,
+            aws_secret_access_key=settings.S3_SECRET_KEY,
+            endpoint_url=settings.S3_ENDPOINT_URL,
+            region_name=settings.S3_REGION_NAME,
+            use_ssl=True,
+            config=Config(signature_version=settings.S3_SIGNATURE_VERSION),
+        )
+        self.bucket = settings.S3_BUCKET_NAME
+        self.key_prefix = key_prefix
+        self.filename = None
+        self.disk_backend = DiskChunkUploadBackend()
+
+    def start(self, filename):
+        self.filename = filename
+        return self.disk_backend.start(filename)
+
+    def upload_chunk(self, chunk):
+        return self.disk_backend.upload_chunk(chunk)
+
+    def abort(self):
+        self.filename = None
+        return self.disk_backend.abort()
+
+    def finish(self):
+        if filepath := self.disk_backend.finish():
+            response = self.s3.upload_file(
+                Filename=filepath,
+                Bucket=settings.S3_BUCKET_NAME,
+                Key=f"{self.key_prefix}{self.filename}",
+            )
+            os.remove(filepath)
+            return response["Location"]