ruạṛ
import os import math import time import requests class MultiPartUpload: """ OCI Object Storage Multipart Upload using a Pre-Authenticated Request (PAR) URL. Supports: - Single PUT for smaller files - Multipart upload for large files (init -> upload parts -> commit) Notes: - OCI multipart parts must be <= 50 GiB. Keep PART_SIZE_BYTES <= 50 * 1024**3 - PAR URL is assumed to be of the form: https://objectstorage.<region>.oraclecloud.com/p/<token>/n/<ns>/b/<bucket>/o/<object> """ def __init__( self, logger, part_size_bytes: int = 25 * 1024**3, # 25 GiB connect_timeout: int = 30, read_timeout: int = 3600, max_retries: int = 5, ): self.logger = logger self.part_size_bytes = part_size_bytes self.connect_timeout = connect_timeout self.read_timeout = read_timeout self.max_retries = max_retries # ---------- Public API ---------- def upload_file(self, local_path: str, par_object_url: str) -> None: self._validate_file(local_path) file_size = self._get_file_size(local_path) if self._should_use_single_put(file_size): self._upload_single_put(local_path, par_object_url) return multipart_base_url = self._initiate_multipart_upload(par_object_url) part_count = self._calculate_part_count(file_size) self.logger.info( f"[OCI MP] Initiated multipart upload: " f"file_size={file_size} bytes, part_size={self.part_size_bytes} bytes, parts={part_count}" ) self._upload_all_parts(local_path, multipart_base_url, file_size, part_count) self._commit_multipart_upload(multipart_base_url) self.logger.info("[OCI MP] Multipart upload committed successfully") # ---------- Validation / Decision ---------- def _validate_file(self, local_path: str) -> None: if not os.path.exists(local_path): raise FileNotFoundError(f"File does not exist: {local_path}") if not os.path.isfile(local_path): raise ValueError(f"Path is not a file: {local_path}") def _get_file_size(self, local_path: str) -> int: size = os.path.getsize(local_path) if size <= 0: raise ValueError(f"Empty file: {local_path}") return size def _should_use_single_put(self, file_size: int) -> bool: return file_size <= self.part_size_bytes # ---------- Single PUT ---------- def _upload_single_put(self, local_path: str, par_object_url: str) -> None: self.logger.info("[OCI] Using single PUT upload") with open(local_path, "rb") as f: r = requests.put( par_object_url, data=f, timeout=self._timeout_tuple(), ) r.raise_for_status() # ---------- Multipart: Init ---------- def _initiate_multipart_upload(self, par_object_url: str) -> str: self.logger.info("[OCI MP] Initiating multipart upload via PAR") r = requests.put( par_object_url, headers={"opc-multipart": "true"}, timeout=self._timeout_tuple(), ) r.raise_for_status() info = r.json() access_uri = info.get("accessUri") if not access_uri: raise RuntimeError(f"Multipart init response missing 'accessUri': {info}") multipart_base_url = self._build_multipart_base_url(par_object_url, access_uri) return multipart_base_url def _build_multipart_base_url(self, par_object_url: str, access_uri: str) -> str: # Everything before "/p/<token>..." is the host root. base_host = par_object_url.split("/p/", 1)[0] return base_host + access_uri # OCI returns accessUri typically ending with "/" # ---------- Multipart: Parts ---------- def _calculate_part_count(self, file_size: int) -> int: return math.ceil(file_size / self.part_size_bytes) def _upload_all_parts(self, local_path: str, multipart_base_url: str, file_size: int, part_count: int) -> None: with open(local_path, "rb") as f: for part_no in range(1, part_count + 1): self._upload_single_part( file_handle=f, multipart_base_url=multipart_base_url, file_size=file_size, part_no=part_no, part_count=part_count, ) def _upload_single_part( self, file_handle, multipart_base_url: str, file_size: int, part_no: int, part_count: int, ) -> None: offset = (part_no - 1) * self.part_size_bytes remaining = file_size - offset part_size = min(self.part_size_bytes, remaining) part_url = self._build_part_url(multipart_base_url, part_no) for attempt in range(self.max_retries): try: self._seek_file(file_handle, offset) reader = self._create_limited_reader(file_handle, part_size) r = requests.put( part_url, data=reader, headers={"Content-Length": str(part_size)}, timeout=self._timeout_tuple(), ) r.raise_for_status() self.logger.info(f"[OCI MP] Uploaded part {part_no}/{part_count} ({part_size} bytes)") return except Exception as e: if attempt == self.max_retries - 1: self.logger.error(f"[OCI MP] Part {part_no} failed after retries: {e}") raise self.logger.warning( f"[OCI MP] Part {part_no} failed (attempt {attempt+1}/{self.max_retries}): {e}" ) self._sleep_backoff(attempt + 1) def _build_part_url(self, multipart_base_url: str, part_no: int) -> str: # OCI expects: PUT {multipart_base_url}{partNumber} return multipart_base_url + str(part_no) def _seek_file(self, file_handle, offset: int) -> None: file_handle.seek(offset) # ---------- Multipart: Commit ---------- def _commit_multipart_upload(self, multipart_base_url: str) -> None: self.logger.info("[OCI MP] Committing multipart upload") r = requests.post( multipart_base_url, timeout=self._timeout_tuple(), ) r.raise_for_status() # ---------- Utils ---------- def _timeout_tuple(self): return (self.connect_timeout, self.read_timeout) def _sleep_backoff(self, attempt: int) -> None: time.sleep(min(2 ** attempt, 30)) def _create_limited_reader(self, file_handle, limit: int): return self._LimitedReader(file_handle, limit) class _LimitedReader: """File-like object that reads at most `limit` bytes from `file_handle`.""" def __init__(self, file_handle, limit: int): self._f = file_handle self._remaining = limit def read(self, size: int = -1) -> bytes: if self._remaining <= 0: return b"" if size < 0 or size > self._remaining: size = self._remaining chunk = self._f.read(size) self._remaining -= len(chunk) return chunk
cải xoăn