1# SPDX-FileCopyrightText: Copyright (c) 2024 NVIDIA CORPORATION & AFFILIATES. All rights reserved.
2# SPDX-License-Identifier: Apache-2.0
3#
4# Licensed under the Apache License, Version 2.0 (the "License");
5# you may not use this file except in compliance with the License.
6# You may obtain a copy of the License at
7#
8# http://www.apache.org/licenses/LICENSE-2.0
9#
10# Unless required by applicable law or agreed to in writing, software
11# distributed under the License is distributed on an "AS IS" BASIS,
12# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
13# See the License for the specific language governing permissions and
14# limitations under the License.
15
16import io
17import os
18import tempfile
19from collections.abc import Callable, Iterator
20from typing import IO, Any, Optional, TypeVar, Union
21
22import oci
23from dateutil.parser import parse as dateutil_parser
24from oci._vendor.requests.exceptions import (
25 ChunkedEncodingError,
26 ConnectionError,
27 ContentDecodingError,
28)
29from oci.exceptions import ServiceError
30from oci.object_storage import ObjectStorageClient, UploadManager
31from oci.retry import DEFAULT_RETRY_STRATEGY, RetryStrategyBuilder
32
33from ..constants import DEFAULT_CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT
34from ..telemetry import Telemetry
35from ..types import (
36 AWARE_DATETIME_MIN,
37 CredentialsProvider,
38 ObjectMetadata,
39 PreconditionFailedError,
40 Range,
41 RetryableError,
42)
43from ..utils import split_path, validate_attributes
44from .base import BaseStorageProvider
45
46_T = TypeVar("_T")
47
48MB = 1024 * 1024
49
50MULTIPART_THRESHOLD = 64 * MB
51MULTIPART_CHUNKSIZE = 32 * MB
52
53PROVIDER = "oci"
54
55
[docs]
56class OracleStorageProvider(BaseStorageProvider):
57 """
58 A concrete implementation of the :py:class:`multistorageclient.types.StorageProvider` for interacting with
59 Oracle Cloud Infrastructure (OCI) Object Storage.
60 """
61
62 def __init__(
63 self,
64 namespace: str,
65 base_path: str = "",
66 credentials_provider: Optional[CredentialsProvider] = None,
67 retry_strategy: Optional[dict[str, Any]] = None,
68 config_dict: Optional[dict[str, Any]] = None,
69 telemetry_provider: Optional[Callable[[], Telemetry]] = None,
70 **kwargs: Any,
71 ) -> None:
72 """
73 Initializes an instance of :py:class:`OracleStorageProvider`.
74
75 :param namespace: The OCI Object Storage namespace. This is a unique identifier assigned to each tenancy.
76 :param base_path: The root prefix path within the bucket where all operations will be scoped.
77 :param credentials_provider: The provider to retrieve OCI credentials.
78 :param retry_strategy: ``oci.retry.RetryStrategyBuilder`` parameters.
79 :param config_dict: Resolved MSC config.
80 :param telemetry_provider: A function that provides a telemetry instance.
81 """
82 super().__init__(
83 base_path=base_path,
84 provider_name=PROVIDER,
85 config_dict=config_dict,
86 telemetry_provider=telemetry_provider,
87 )
88
89 self._namespace = namespace
90 self._credentials_provider = credentials_provider
91 self._retry_strategy = (
92 DEFAULT_RETRY_STRATEGY
93 if retry_strategy is None
94 else RetryStrategyBuilder(**retry_strategy).get_retry_strategy()
95 )
96 self._timeout = kwargs.get("timeout")
97 if self._timeout is None:
98 self._timeout = (DEFAULT_CONNECT_TIMEOUT, DEFAULT_READ_TIMEOUT)
99 self._oci_client = self._create_oci_client()
100 self._upload_manager = UploadManager(self._oci_client)
101 self._multipart_threshold = int(kwargs.get("multipart_threshold", MULTIPART_THRESHOLD))
102 self._multipart_chunksize = int(kwargs.get("multipart_chunksize", MULTIPART_CHUNKSIZE))
103
104 def _create_oci_client(self) -> ObjectStorageClient:
105 config = oci.config.from_file()
106 client = ObjectStorageClient(config, retry_strategy=self._retry_strategy)
107 client.base_client.timeout = self._timeout
108 return client
109
110 def _refresh_oci_client_if_needed(self) -> None:
111 """
112 Refreshes the OCI client if the current credentials are expired.
113 """
114 if self._credentials_provider:
115 credentials = self._credentials_provider.get_credentials()
116 if credentials.is_expired():
117 self._credentials_provider.refresh_credentials()
118 self._oci_client = self._create_oci_client()
119 self._upload_manager = UploadManager(
120 self._oci_client, allow_parallel_uploads=True, parallel_process_count=4
121 )
122
123 def _translate_errors(
124 self,
125 func: Callable[[], _T],
126 operation: str,
127 bucket: str,
128 key: str,
129 ) -> _T:
130 """
131 Translates errors like timeouts and client errors.
132
133 :param func: The function that performs the actual object storage operation.
134 :param operation: The type of operation being performed (e.g., "PUT", "GET", "DELETE").
135 :param bucket: The name of the object storage bucket involved in the operation.
136 :param key: The key of the object within the object storage bucket.
137
138 :return: The result of the object storage operation, typically the return value of the `func` callable.
139 """
140 try:
141 return func()
142 except ServiceError as error:
143 status_code = error.status
144 request_id = error.request_id
145 endpoint = error.request_endpoint
146 error_info = f"request_id: {request_id}, endpoint: {endpoint}, status_code: {status_code}"
147
148 if status_code == 404:
149 raise FileNotFoundError(f"Object {bucket}/{key} does not exist. {error_info}") # pylint: disable=raise-missing-from
150 elif status_code == 412:
151 raise PreconditionFailedError(
152 f"Failed to {operation} object(s) at {bucket}/{key}. {error_info}"
153 ) from error
154 elif status_code == 429:
155 raise RetryableError(
156 f"Too many request to {operation} object(s) at {bucket}/{key}. {error_info}"
157 ) from error
158 else:
159 raise RuntimeError(f"Failed to {operation} object(s) at {bucket}/{key}. {error_info}") from error
160 except (ConnectionError, ChunkedEncodingError, ContentDecodingError) as error:
161 raise RetryableError(
162 f"Failed to {operation} object(s) at {bucket}/{key}, error type: {type(error).__name__}"
163 ) from error
164 except FileNotFoundError:
165 raise
166 except Exception as error:
167 raise RuntimeError(
168 f"Failed to {operation} object(s) at {bucket}/{key}, error type: {type(error).__name__}, error: {error}"
169 ) from error
170
171 def _put_object(
172 self,
173 path: str,
174 body: bytes,
175 if_match: Optional[str] = None,
176 if_none_match: Optional[str] = None,
177 attributes: Optional[dict[str, str]] = None,
178 ) -> int:
179 bucket, key = split_path(path)
180 self._refresh_oci_client_if_needed()
181
182 # OCI only supports if_none_match=="*"
183 # refer: https://docs.oracle.com/en-us/iaas/tools/python/2.150.0/api/object_storage/client/oci.object_storage.ObjectStorageClient.html?highlight=put_object#oci.object_storage.ObjectStorageClient.put_object
184 def _invoke_api() -> int:
185 validated_attributes = validate_attributes(attributes)
186 self._oci_client.put_object(
187 namespace_name=self._namespace,
188 bucket_name=bucket,
189 object_name=key,
190 put_object_body=body,
191 opc_meta=validated_attributes or {}, # Pass metadata or empty dict
192 if_match=if_match,
193 if_none_match=if_none_match,
194 )
195
196 return len(body)
197
198 return self._translate_errors(_invoke_api, operation="PUT", bucket=bucket, key=key)
199
200 def _get_object(self, path: str, byte_range: Optional[Range] = None) -> bytes:
201 bucket, key = split_path(path)
202 self._refresh_oci_client_if_needed()
203
204 def _invoke_api() -> bytes:
205 if byte_range:
206 bytes_range = f"bytes={byte_range.offset}-{byte_range.offset + byte_range.size - 1}"
207 else:
208 bytes_range = None
209 response = self._oci_client.get_object(
210 namespace_name=self._namespace, bucket_name=bucket, object_name=key, range=bytes_range
211 )
212 return response.data.content # pyright: ignore [reportOptionalMemberAccess]
213
214 return self._translate_errors(_invoke_api, operation="GET", bucket=bucket, key=key)
215
216 def _copy_object(self, src_path: str, dest_path: str) -> int:
217 src_bucket, src_key = split_path(src_path)
218 dest_bucket, dest_key = split_path(dest_path)
219 self._refresh_oci_client_if_needed()
220
221 src_object = self._get_object_metadata(src_path)
222
223 def _invoke_api() -> int:
224 copy_details = oci.object_storage.models.CopyObjectDetails(
225 source_object_name=src_key, destination_bucket=dest_bucket, destination_object_name=dest_key
226 )
227
228 self._oci_client.copy_object(
229 namespace_name=self._namespace, bucket_name=src_bucket, copy_object_details=copy_details
230 )
231
232 return src_object.content_length
233
234 return self._translate_errors(_invoke_api, operation="COPY", bucket=src_bucket, key=src_key)
235
236 def _delete_object(self, path: str, if_match: Optional[str] = None) -> None:
237 bucket, key = split_path(path)
238 self._refresh_oci_client_if_needed()
239
240 def _invoke_api() -> None:
241 namespace_name = self._namespace
242 bucket_name = bucket
243 object_name = key
244 if if_match is not None:
245 self._oci_client.delete_object(namespace_name, bucket_name, object_name, if_match=if_match)
246 else:
247 self._oci_client.delete_object(namespace_name, bucket_name, object_name)
248
249 return self._translate_errors(_invoke_api, operation="DELETE", bucket=bucket, key=key)
250
251 def _is_dir(self, path: str) -> bool:
252 # Ensure the path ends with '/' to mimic a directory
253 path = self._append_delimiter(path)
254
255 bucket, key = split_path(path)
256 self._refresh_oci_client_if_needed()
257
258 def _invoke_api() -> bool:
259 # List objects with the given prefix
260 response = self._oci_client.list_objects(
261 namespace_name=self._namespace,
262 bucket_name=bucket,
263 prefix=key,
264 delimiter="/",
265 )
266 # Check if there are any contents or common prefixes
267 if response:
268 return bool(response.data.objects or response.data.prefixes)
269 return False
270
271 return self._translate_errors(_invoke_api, operation="LIST", bucket=bucket, key=key)
272
273 def _get_object_metadata(self, path: str, strict: bool = True) -> ObjectMetadata:
274 bucket, key = split_path(path)
275 if path.endswith("/") or (bucket and not key):
276 # If path ends with "/" or empty key name is provided, then assume it's a "directory",
277 # which metadata is not guaranteed to exist for cases such as
278 # "virtual prefix" that was never explicitly created.
279 if self._is_dir(path):
280 return ObjectMetadata(
281 key=path,
282 type="directory",
283 content_length=0,
284 last_modified=AWARE_DATETIME_MIN,
285 )
286 else:
287 raise FileNotFoundError(f"Directory {path} does not exist.")
288 else:
289 self._refresh_oci_client_if_needed()
290
291 def _invoke_api() -> ObjectMetadata:
292 response = self._oci_client.head_object(
293 namespace_name=self._namespace, bucket_name=bucket, object_name=key
294 )
295
296 # Extract custom metadata from headers with 'opc-meta-' prefix
297 attributes = {}
298 if response.headers: # pyright: ignore [reportOptionalMemberAccess]
299 for metadata_key, metadata_val in response.headers.items(): # pyright: ignore [reportOptionalMemberAccess]
300 if metadata_key.startswith("opc-meta-"):
301 # Remove the 'opc-meta-' prefix to get the original key
302 metadata_key = metadata_key[len("opc-meta-") :]
303 attributes[metadata_key] = metadata_val
304
305 return ObjectMetadata(
306 key=path,
307 content_length=int(response.headers["Content-Length"]), # pyright: ignore [reportOptionalMemberAccess]
308 content_type=response.headers.get("Content-Type", None), # pyright: ignore [reportOptionalMemberAccess]
309 last_modified=dateutil_parser(response.headers["last-modified"]), # pyright: ignore [reportOptionalMemberAccess]
310 etag=response.headers.get("etag", None), # pyright: ignore [reportOptionalMemberAccess]
311 metadata=attributes if attributes else None,
312 )
313
314 try:
315 return self._translate_errors(_invoke_api, operation="HEAD", bucket=bucket, key=key)
316 except FileNotFoundError as error:
317 if strict:
318 # If the object does not exist on the given path, we will append a trailing slash and
319 # check if the path is a directory.
320 path = self._append_delimiter(path)
321 if self._is_dir(path):
322 return ObjectMetadata(
323 key=path,
324 type="directory",
325 content_length=0,
326 last_modified=AWARE_DATETIME_MIN,
327 )
328 raise error
329
330 def _list_objects(
331 self,
332 path: str,
333 start_after: Optional[str] = None,
334 end_at: Optional[str] = None,
335 include_directories: bool = False,
336 follow_symlinks: bool = True,
337 ) -> Iterator[ObjectMetadata]:
338 bucket, prefix = split_path(path)
339 self._refresh_oci_client_if_needed()
340
341 def _invoke_api() -> Iterator[ObjectMetadata]:
342 # ListObjects only includes object names by default.
343 #
344 # Request additional fields needed for creating an ObjectMetadata.
345 fields = ",".join(
346 [
347 "etag",
348 "name",
349 "size",
350 "timeModified",
351 ]
352 )
353 next_start_with: Optional[str] = start_after
354 while True:
355 if include_directories:
356 response = self._oci_client.list_objects(
357 namespace_name=self._namespace,
358 bucket_name=bucket,
359 prefix=prefix,
360 # This is ≥ instead of >.
361 start=next_start_with,
362 delimiter="/",
363 fields=fields,
364 )
365 else:
366 response = self._oci_client.list_objects(
367 namespace_name=self._namespace,
368 bucket_name=bucket,
369 prefix=prefix,
370 # This is ≥ instead of >.
371 start=next_start_with,
372 fields=fields,
373 )
374
375 if not response:
376 return []
377
378 if include_directories:
379 for directory in response.data.prefixes:
380 yield ObjectMetadata(
381 key=directory.rstrip("/"),
382 type="directory",
383 content_length=0,
384 last_modified=AWARE_DATETIME_MIN,
385 )
386
387 # OCI guarantees lexicographical order.
388 for response_object in response.data.objects: # pyright: ignore [reportOptionalMemberAccess]
389 key = response_object.name
390 if (start_after is None or start_after < key) and (end_at is None or key <= end_at):
391 if key.endswith("/"):
392 if include_directories:
393 yield ObjectMetadata(
394 key=os.path.join(bucket, key.rstrip("/")),
395 type="directory",
396 content_length=0,
397 last_modified=response_object.time_modified,
398 )
399 else:
400 yield ObjectMetadata(
401 key=os.path.join(bucket, key),
402 type="file",
403 content_length=response_object.size,
404 last_modified=response_object.time_modified,
405 etag=response_object.etag,
406 )
407 elif start_after != key:
408 return
409 next_start_with = response.data.next_start_with # pyright: ignore [reportOptionalMemberAccess]
410 if next_start_with is None or (end_at is not None and end_at < next_start_with):
411 return
412
413 return self._translate_errors(_invoke_api, operation="LIST", bucket=bucket, key=prefix)
414
415 def _upload_file(self, remote_path: str, f: Union[str, IO], attributes: Optional[dict[str, str]] = None) -> int:
416 bucket, key = split_path(remote_path)
417 file_size: int = 0
418 self._refresh_oci_client_if_needed()
419
420 validated_attributes = validate_attributes(attributes)
421 if isinstance(f, str):
422 file_size = os.path.getsize(f)
423
424 def _invoke_api() -> int:
425 if file_size > self._multipart_threshold:
426 self._upload_manager.upload_file(
427 namespace_name=self._namespace,
428 bucket_name=bucket,
429 object_name=key,
430 file_path=f,
431 part_size=self._multipart_chunksize,
432 allow_parallel_uploads=True,
433 metadata=validated_attributes or {},
434 )
435 else:
436 self._upload_manager.upload_file(
437 namespace_name=self._namespace,
438 bucket_name=bucket,
439 object_name=key,
440 file_path=f,
441 metadata=validated_attributes or {},
442 )
443
444 return file_size
445
446 return self._translate_errors(_invoke_api, operation="PUT", bucket=bucket, key=key)
447 else:
448 # Convert file-like object to BytesIO because stream_ref cannot work with StringIO.
449 if isinstance(f, io.StringIO):
450 f = io.BytesIO(f.getvalue().encode("utf-8"))
451
452 f.seek(0, io.SEEK_END)
453 file_size = f.tell()
454 f.seek(0)
455
456 def _invoke_api() -> int:
457 if file_size > self._multipart_threshold:
458 self._upload_manager.upload_stream(
459 namespace_name=self._namespace,
460 bucket_name=bucket,
461 object_name=key,
462 stream_ref=f,
463 part_size=self._multipart_chunksize,
464 allow_parallel_uploads=True,
465 metadata=validated_attributes or {},
466 )
467 else:
468 self._upload_manager.upload_stream(
469 namespace_name=self._namespace,
470 bucket_name=bucket,
471 object_name=key,
472 stream_ref=f,
473 metadata=validated_attributes or {},
474 )
475
476 return file_size
477
478 return self._translate_errors(_invoke_api, operation="PUT", bucket=bucket, key=key)
479
480 def _download_file(self, remote_path: str, f: Union[str, IO], metadata: Optional[ObjectMetadata] = None) -> int:
481 self._refresh_oci_client_if_needed()
482
483 if metadata is None:
484 metadata = self._get_object_metadata(remote_path)
485
486 bucket, key = split_path(remote_path)
487
488 if isinstance(f, str):
489 if os.path.dirname(f):
490 os.makedirs(os.path.dirname(f), exist_ok=True)
491
492 def _invoke_api() -> int:
493 response = self._oci_client.get_object(
494 namespace_name=self._namespace, bucket_name=bucket, object_name=key
495 )
496 with tempfile.NamedTemporaryFile(mode="wb", delete=False, dir=os.path.dirname(f), prefix=".") as fp:
497 temp_file_path = fp.name
498 for chunk in response.data.raw.stream(1024 * 1024, decode_content=False): # pyright: ignore [reportOptionalMemberAccess]
499 fp.write(chunk)
500 os.rename(src=temp_file_path, dst=f)
501
502 return metadata.content_length
503
504 return self._translate_errors(_invoke_api, operation="GET", bucket=bucket, key=key)
505 else:
506
507 def _invoke_api() -> int:
508 response = self._oci_client.get_object(
509 namespace_name=self._namespace, bucket_name=bucket, object_name=key
510 )
511 # Convert file-like object to BytesIO because stream_ref cannot work with StringIO.
512 if isinstance(f, io.StringIO):
513 bytes_fileobj = io.BytesIO()
514 for chunk in response.data.raw.stream(1024 * 1024, decode_content=False): # pyright: ignore [reportOptionalMemberAccess]
515 bytes_fileobj.write(chunk)
516 f.write(bytes_fileobj.getvalue().decode("utf-8"))
517 else:
518 for chunk in response.data.raw.stream(1024 * 1024, decode_content=False): # pyright: ignore [reportOptionalMemberAccess]
519 f.write(chunk)
520
521 return metadata.content_length
522
523 return self._translate_errors(_invoke_api, operation="GET", bucket=bucket, key=key)