Oraclecloud SDK Patterns

import oci
import threading

class OCIClients:
    """Thread-safe singleton for OCI service clients.

    Prevents the Instance Principal memory leak by reusing clients
    instead of creating new ones per request.
    """
    _lock = threading.Lock()
    _instance = None

    def __init__(self):
        self._config = oci.config.from_file("~/.oci/config")
        oci.config.validate_config(self._config)

        # Create clients once — reuse everywhere
        self._compute = None
        self._network = None
        self._object_storage = None
        self._identity = None

    @classmethod
    def get(cls):
        if cls._instance is None:
            with cls._lock:
                if cls._instance is None:
                    cls._instance = cls()
        return cls._instance

    @property
    def config(self):
        return self._config

    @property
    def compute(self):
        if self._compute is None:
            self._compute = oci.core.ComputeClient(
                self._config, retry_strategy=oci.retry.DEFAULT_RETRY_STRATEGY
            )
        return self._compute

    @property
    def network(self):
        if self._network is None:
            self._network = oci.core.VirtualNetworkClient(
                self._config, retry_strategy=oci.retry.DEFAULT_RETRY_STRATEGY
            )
        return self._network

    @property
    def object_storage(self):
        if self._object_storage is None:
            self._object_storage = oci.object_storage.ObjectStorageClient(
                self._config, retry_strategy=oci.retry.DEFAULT_RETRY_STRATEGY
            )
        return self._object_storage

    @property
    def identity(self):
        if self._identity is None:
            self._identity = oci.identity.IdentityClient(
                self._config, retry_strategy=oci.retry.DEFAULT_RETRY_STRATEGY
            )
        return self._identity


# Usage: never create clients directly
clients = OCIClients.get()
instances = clients.compute.list_instances(
    compartment_id=clients.config["tenancy"]
)

import oci

config = oci.config.from_file("~/.oci/config")

# Compute: 10s connect, 60s read
compute = oci.core.ComputeClient(
    config,
    timeout=(10, 60)  # (connect_timeout, read_timeout) in seconds
)

# Object Storage: 10s connect, 300s read (large uploads)
object_storage = oci.object_storage.ObjectStorageClient(
    config,
    timeout=(10, 300)
)

# Database: 10s connect, 120s read (long queries)
database = oci.database.DatabaseClient(
    config,
    timeout=(10, 120)
)

import oci

custom_retry = oci.retry.RetryStrategyBuilder(
    max_attempts_check=True,
    max_attempts=5,
    total_elapsed_time_check=True,
    total_elapsed_time_seconds=300,
    retry_max_wait_between_calls_seconds=30,
    retry_base_sleep_time_seconds=2,
    service_error_check=True,
    service_error_retry_on_any_5xx=True,
    service_error_retry_config={
        429: []  # Retry on all 429 errors (no Retry-After header in OCI)
    },
    backoff_type=oci.retry.BACKOFF_DECORRELATED_JITTER
).get_retry_strategy()

compute = oci.core.ComputeClient(config, retry_strategy=custom_retry)

import time
import random
import oci

def call_with_retry(fn, max_retries=5, base_delay=2):
    """Execute an OCI SDK call with exponential backoff.

    Retries on: 429 TooManyRequests, 500 InternalError, -1 timeout.
    Raises immediately on: 401, 404, 400.
    """
    for attempt in range(max_retries):
        try:
            return fn()
        except oci.exceptions.ServiceError as e:
            if e.status in (429, 500, 502, 503, 504) or e.status == -1:
                delay = base_delay * (2 ** attempt) + random.uniform(0, 1)
                print(f"Attempt {attempt + 1} failed ({e.status}). Retry in {delay:.1f}s")
                time.sleep(delay)
            else:
                raise  # 401, 404, 400 — don't retry
    raise RuntimeError(f"Failed after {max_retries} retries")

# Usage
config = oci.config.from_file("~/.oci/config")
compute = oci.core.ComputeClient(config)

instances = call_with_retry(
    lambda: compute.list_instances(compartment_id=config["tenancy"])
)

import oci

config = oci.config.from_file("~/.oci/config")
compute = oci.core.ComputeClient(config)

# Automatic pagination — returns ALL instances
all_instances = oci.pagination.list_call_get_all_results(
    compute.list_instances,
    compartment_id=config["tenancy"]
).data

print(f"Total instances: {len(all_instances)}")

# Lazy pagination — yields pages (memory-efficient for large datasets)
for page in oci.pagination.list_call_get_all_results_generator(
    compute.list_instances,
    "response",
    compartment_id=config["tenancy"]
):
    for inst in page.data:
        print(f"{inst.display_name}: {inst.lifecycle_state}")

import oci

config = oci.config.from_file("~/.oci/config")
compute = oci.core.ComputeClient(config)
compute_composite = oci.core.ComputeClientCompositeOperations(compute)

# Launch and wait for RUNNING state
launch_details = oci.core.models.LaunchInstanceDetails(
    compartment_id=config["tenancy"],
    availability_domain="Uocm:US-ASHBURN-AD-1",
    display_name="sdk-pattern-demo",
    shape="VM.Standard.E4.Flex",
    shape_config=oci.core.models.LaunchInstanceShapeConfigDetails(
        ocpus=1, memory_in_gbs=8
    ),
    source_details=oci.core.models.InstanceSourceViaImageDetails(
        image_id="ocid1.image.oc1.iad.aaaa..."
    ),
    create_vnic_details=oci.core.models.CreateVnicDetails(
        subnet_id="ocid1.subnet.oc1.iad.aaaa..."
    )
)

response = compute_composite.launch_instance_and_wait_for_state(
    launch_details,
    wait_for_states=[
        oci.core.models.Instance.LIFECYCLE_STATE_RUNNING
    ]
)
print(f"Instance running: {response.data.id}")