Oci Sdk

Oracle Cloud Infrastructure (OCI) SDK Integration Patterns

Integrates Oracle Cloud Infrastructure services using the OCI Python SDK. Covers config-based authentication, Compute instance management, Object Storage CRUD, Autonomous Database operations, and Functions (FaaS) invocation with OCI-specific patterns for pagination, tagging, and composite operations.

TL;DR Checklist

• Use oci.config.from_file() to load the SDK config — never hardcode credentials in source
• Initialize service clients with the config dict: oci.core.ComputeClient(config)
• Handle oci.exceptions.ServiceError with specific HTTP status codes
• Use composite operations (ComputeClientCompositeOperations) for multi-step provisioning
• Use pagination with has_next_page / next_page for list operations
• Use resource principals for OCI Functions and OKE workloads (no config file needed)

When to Use

Use this skill when:

• Provisioning and managing Compute instances, VCNs, and Block Volumes from Python
• Building applications that read/write data to Object Storage buckets
• Managing Autonomous Database instances (ADW/ATP) programmatically
• Deploying and invoking OCI Functions (Fn Project-based serverless)
• Automating OCI resource lifecycle with Python scripts
• Implementing multi-compartment resource management with tagging

When NOT to Use

• For declarative infrastructure management (use Terraform or Resource Manager stacks)
• For one-off console operations (use OCI CLI or Console)
• When you need cross-cloud orchestration (use multi-cloud abstraction libraries)
• For simple Object Storage operations from the CLI (use oci os CLI commands)

Core Workflow

1. Load Configuration and Initialize Client

OCI SDK uses a config file (~/.oci/config) with profile-based authentication.

import oci
from oci.config import validate_config
from oci.exceptions import ServiceError, ConfigFileNotFound

try:
    # Load configuration from default location (~/.oci/config)
    config = oci.config.from_file(
        file_location="~/.oci/config",
        profile_name="DEFAULT",  # Use named profiles for different environments
    )
    validate_config(config)

except ConfigFileNotFound as err:
    raise RuntimeError(
        "OCI config not found. Create ~/.oci/config with your API key."
    ) from err

# Initialize service clients
identity_client = oci.identity.IdentityClient(config)
compute_client = oci.core.ComputeClient(config)
storage_client = oci.object_storage.ObjectStorageClient(config)

# Get the tenancy OCID for compartment-scoped operations
tenancy_id = config["tenancy"]

Checkpoint: Verify authentication by calling identity_client.get_user(config["user"]). A successful response returns user details. Catch ServiceError with status 401 for auth failures.

2. Compute Instance Management

def launch_instance(
    config: dict,
    compartment_id: str,
    availability_domain: str,
    subnet_id: str,
    image_id: str,
    instance_name: str = "py-sdk-instance",
) -> oci.core.models.Instance:
    """Launch a compute instance with specified resources."""
    compute_client = oci.core.ComputeClient(config)
    compute_ops = oci.core.ComputeClientCompositeOperations(compute_client)

    # Define instance launch details
    instance_details = oci.core.models.LaunchInstanceDetails(
        compartment_id=compartment_id,
        availability_domain=availability_domain,
        display_name=instance_name,
        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=image_id,
        ),
        create_vnic_details=oci.core.models.CreateVnicDetails(
            subnet_id=subnet_id,
            assign_public_ip=True,
        ),
    )

    try:
        # Composite operation: launch + wait for RUNNING state
        response = compute_ops.launch_instance_and_wait_for_state(
            instance_details,
            wait_for_states=[oci.core.models.Instance.LIFECYCLE_STATE_RUNNING],
        )
        return response.data

    except ServiceError as err:
        if err.status == 429:
            raise RuntimeError("Rate limited — retry with backoff") from err
        if err.status == 401:
            raise PermissionError("Invalid OCI credentials") from err
        raise

Checkpoint: Instance launch is an LRO. The composite operation blocks until the instance reaches RUNNING. Set wait_for_states to control which state is sufficient for your use case.

3. Object Storage CRUD

from oci.object_storage.models import CreateBucketDetails


class ObjectStorageManager:
    """Manage OCI Object Storage buckets and objects."""

    def __init__(self, config: dict):
        self.client = oci.object_storage.ObjectStorageClient(config)
        self.namespace = self.client.get_namespace().data

    def create_bucket(
        self, compartment_id: str, bucket_name: str
    ) -> oci.object_storage.models.Bucket:
        """Create a standard Object Storage bucket."""
        try:
            details = CreateBucketDetails(
                compartment_id=compartment_id,
                name=bucket_name,
                public_access_type="NoPublicAccess",
                storage_tier="Standard",
            )
            response = self.client.create_bucket(
                namespace_name=self.namespace,
                create_bucket_details=details,
            )
            return response.data
        except ServiceError as err:
            if err.status == 409 and "already exists" in err.message:
                # Idempotent: bucket exists
                return self.get_bucket(bucket_name)
            raise

    def get_bucket(self, bucket_name: str) -> oci.object_storage.models.Bucket:
        """Get bucket metadata."""
        response = self.client.get_bucket(
            namespace_name=self.namespace,
            bucket_name=bucket_name,
        )
        return response.data

    def upload_object(
        self, bucket_name: str, object_name: str, data: bytes
    ) -> str:
        """Upload an object. Returns the object's OCID."""
        try:
            response = self.client.put_object(
                namespace_name=self.namespace,
                bucket_name=bucket_name,
                object_name=object_name,
                put_object_body=data,
            )
            return (
                f"/n/{self.namespace}/b/{bucket_name}/o/{object_name}"
            )
        except ServiceError as err:
            raise RuntimeError(
                f"Failed to upload {object_name} to bucket {bucket_name}"
            ) from err

    def list_objects(self, bucket_name: str, prefix: str = "") -> list[str]:
        """List object names with optional prefix."""
        response = self.client.list_objects(
            namespace_name=self.namespace,
            bucket_name=bucket_name,
            prefix=prefix,
        )
        return [obj.name for obj in response.data.objects]

Checkpoint: Object Storage namespace is the tenancy's unique identifier (not configurable). Use get_namespace() to discover it rather than hardcoding.

4. Pagination for List Operations

OCI list operations return paginated results. Use the has_next_page and next_page pattern.

def list_all_compartments(identity_client, tenancy_id: str) -> list[dict]:
    """List all compartments with pagination handling."""
    all_compartments = []
    page = None

    while True:
        response = identity_client.list_compartments(
            compartment_id=tenancy_id,
            page=page,
            limit=50,
        )
        all_compartments.extend(response.data)

        if response.has_next_page:
            page = response.next_page
        else:
            break

    return [
        {"id": c.id, "name": c.name, "state": c.lifecycle_state}
        for c in all_compartments
    ]

Implementation Patterns

Pattern 1: Autonomous Database Operations

def create_autonomous_database(
    config: dict,
    compartment_id: str,
    db_name: str,
    admin_password: str,
) -> oci.database.models.AutonomousDatabase:
    """Create an Autonomous Database (ADW or ATP)."""
    db_client = oci.database.DatabaseClient(config)
    db_ops = oci.database.DatabaseClientCompositeOperations(db_client)

    details = oci.database.models.CreateAutonomousDatabaseDetails(
        compartment_id=compartment_id,
        db_name=db_name,
        display_name=f"{db_name}-adw",
        admin_password=admin_password,
        data_storage_size_in_tbs=1,
        cpu_core_count=1,
        db_workload="DW",  # "DW" for ADW, "OLTP" for ATP
        is_auto_scaling_enabled=True,
        license_model="LICENSE_INCLUDED",
    )

    try:
        response = db_ops.create_autonomous_database_and_wait_for_state(
            details,
            wait_for_states=[
                oci.database.models.AutonomousDatabase.LIFECYCLE_STATE_AVAILABLE,
            ],
        )
        return response.data
    except ServiceError as err:
        raise RuntimeError(
            f"Failed to create Autonomous DB: {err.message}"
        ) from err

Pattern 2: Invoke OCI Function

import json


def invoke_function(
    config: dict,
    function_id: str,
    payload: dict,
) -> dict:
    """Invoke an OCI Function with JSON payload."""
    # Functions Invoke client requires specific endpoint from the function
    fn_client = oci.functions.FunctionsManagementClient(config)

    # Get function details for invoke endpoint
    fn = fn_client.get_function(function_id).data

    # Create invocation client with function's invoke endpoint
    invoke_client = oci.functions.FunctionsInvokeClient(
        config,
        service_endpoint=fn.invoke_endpoint,
    )

    try:
        response = invoke_client.invoke_function(
            function_id=function_id,
            invoke_function_body=json.dumps(payload),
        )
        return json.loads(response.data.text)
    except ServiceError as err:
        raise RuntimeError(
            f"Function invocation failed: {err.message}"
        ) from err

BAD vs GOOD: Config Handling

# ❌ BAD — Hardcoded credentials and manual client setup without error handling
import oci
config = {
    "user": "ocid1.user.oc1..aaaa...",
    "key_file": "/home/user/oci_key.pem",
    "tenancy": "ocid1.tenancy.oc1...",
    "region": "us-ashburn-1",
    "fingerprint": "12:34:56:78:90:ab:cd:ef:...",
}
identity = oci.identity.IdentityClient(config)
user = identity.get_user(config["user"])
print(user.data)

# ✅ GOOD — Config file with validation and error handling
from oci.config import from_file, validate_config
from oci.exceptions import ServiceError, ConfigFileNotFound

try:
    config = from_file()
    validate_config(config)
except ConfigFileNotFound:
    raise RuntimeError("Create ~/.oci/config with your API key")

identity = oci.identity.IdentityClient(config)
try:
    user = identity.get_user(config["user"]).data
    print(f"Authenticated as: {user.name}")
except ServiceError as err:
    if err.status == 401:
        raise PermissionError("Invalid OCI credentials — check key/fingerprint")
    raise

BAD vs GOOD: Object Storage Upload

# ❌ BAD — No namespace discovery, no error handling
storage = oci.object_storage.ObjectStorageClient(config)
storage.put_object(
    namespace_name="ax1b2c3d4e5f",
    bucket_name="my-bucket",
    object_name="data.json",
    put_object_body='{"key": "value"}',
)

# ✅ GOOD — Namespace discovery + structured error handling
def upload_config(config: dict, bucket_name: str, data: bytes) -> str:
    storage = oci.object_storage.ObjectStorageClient(config)
    try:
        namespace = storage.get_namespace().data
    except ServiceError as err:
        raise RuntimeError("Cannot discover Object Storage namespace") from err

    try:
        storage.put_object(
            namespace_name=namespace,
            bucket_name=bucket_name,
            object_name="config.json",
            put_object_body=data,
        )
        return f"/n/{namespace}/b/{bucket_name}/o/config.json"
    except ServiceError as err:
        if err.status == 404:
            raise FileNotFoundError(f"Bucket '{bucket_name}' not found") from err
        raise

Constraints

MUST DO

• Load config from ~/.oci/config using oci.config.from_file() — never hardcode user OCID, key, or tenancy
• Use validate_config() after loading to catch common configuration errors early
• Use composite operations (*ClientCompositeOperations) for resource creation that needs state waiting
• Handle ServiceError with specific status codes: 401 (auth), 404 (not found), 409 (conflict), 429 (rate limit)
• Use pagination with has_next_page / next_page for list operations (>100 results)
• Use resource principals for OCI Functions and OKE workloads where config files are unavailable

MUST NOT DO

• Hardcode OCIDs, private key paths, or tenancy identifiers in application source code
• Ignore pagination on list operations — OCI defaults to 50-100 results per page
• Use synchronous patterns for long-running operations without composite operations
• Share API key private keys across environments — use separate keys per environment
• Assume Object Storage namespace — always discover it via get_namespace()

Output Template

When implementing an OCI SDK integration, structure your output as:

1. Config Loading — oci.config.from_file() with validation
2. Service Client Init — Per-service client from config dict
3. Operation — Resource creation with composite ops for LROs
4. Error Handling — ServiceError with status code branching
5. Pagination — while loop with has_next_page for list operations
6. Return Value — Parsed .data attribute from response objects

Related Skills

Live References

• OCI Python SDK Documentation — Official SDK reference
• OCI Python SDK Quickstart — Getting started guide
• OCI Python SDK GitHub — Source code and examples
• OCI Python SDK Examples — Runable example scripts
• OCI Config File Format — SDK config file reference
• OCI API Key Management — How to create API keys
• OCI Resource Principal — Auth for OKE/Functions