"""
Tools for managing additional OCI resources.
"""
import logging
from typing import Dict, List, Any, Optional
import oci
logger = logging.getLogger(__name__)
def list_availability_domains(identity_client: oci.identity.IdentityClient, compartment_id: str) -> List[Dict[str, Any]]:
"""
List all availability domains in a compartment.
Args:
identity_client: OCI Identity client
compartment_id: OCID of the compartment
Returns:
List of availability domains with their details
"""
try:
ads_response = identity_client.list_availability_domains(compartment_id)
ads = []
for ad in ads_response.data:
ads.append({
"name": ad.name,
"id": ad.id,
"compartment_id": ad.compartment_id,
})
logger.info(f"Found {len(ads)} availability domains in compartment {compartment_id}")
return ads
except Exception as e:
logger.exception(f"Error listing availability domains: {e}")
raise
def list_fault_domains(identity_client: oci.identity.IdentityClient, compartment_id: str,
availability_domain: str) -> List[Dict[str, Any]]:
"""
List all fault domains in an availability domain.
Args:
identity_client: OCI Identity client
compartment_id: OCID of the compartment
availability_domain: Name of the availability domain
Returns:
List of fault domains with their details
"""
try:
fds_response = identity_client.list_fault_domains(
compartment_id=compartment_id,
availability_domain=availability_domain
)
fds = []
for fd in fds_response.data:
fds.append({
"name": fd.name,
"id": fd.id,
"compartment_id": fd.compartment_id,
"availability_domain": fd.availability_domain,
})
logger.info(f"Found {len(fds)} fault domains in availability domain {availability_domain}")
return fds
except Exception as e:
logger.exception(f"Error listing fault domains: {e}")
raise
def list_images(compute_client: oci.core.ComputeClient, compartment_id: str) -> List[Dict[str, Any]]:
"""
List all images in a compartment.
Args:
compute_client: OCI Compute client
compartment_id: OCID of the compartment
Returns:
List of images with their details
"""
try:
images_response = oci.pagination.list_call_get_all_results(
compute_client.list_images,
compartment_id
)
images = []
for image in images_response.data:
images.append({
"id": image.id,
"display_name": image.display_name,
"compartment_id": image.compartment_id,
"operating_system": image.operating_system,
"operating_system_version": image.operating_system_version,
"lifecycle_state": image.lifecycle_state,
"time_created": str(image.time_created),
"size_in_mbs": image.size_in_mbs,
"base_image_id": image.base_image_id,
"create_image_allowed": image.create_image_allowed,
"listing_type": image.listing_type,
})
logger.info(f"Found {len(images)} images in compartment {compartment_id}")
return images
except Exception as e:
logger.exception(f"Error listing images: {e}")
raise
def get_image(compute_client: oci.core.ComputeClient, image_id: str) -> Dict[str, Any]:
"""
Get details of a specific image.
Args:
compute_client: OCI Compute client
image_id: OCID of the image
Returns:
Details of the image
"""
try:
image = compute_client.get_image(image_id).data
image_details = {
"id": image.id,
"display_name": image.display_name,
"compartment_id": image.compartment_id,
"operating_system": image.operating_system,
"operating_system_version": image.operating_system_version,
"lifecycle_state": image.lifecycle_state,
"time_created": str(image.time_created),
"size_in_mbs": image.size_in_mbs,
"base_image_id": image.base_image_id,
"create_image_allowed": image.create_image_allowed,
"listing_type": image.listing_type,
"launch_mode": image.launch_mode,
"launch_options": {
"boot_volume_type": image.launch_options.boot_volume_type if image.launch_options else None,
"firmware": image.launch_options.firmware if image.launch_options else None,
"network_type": image.launch_options.network_type if image.launch_options else None,
"remote_data_volume_type": image.launch_options.remote_data_volume_type if image.launch_options else None,
"is_pv_encryption_in_transit_enabled": image.launch_options.is_pv_encryption_in_transit_enabled if image.launch_options else None,
"is_consistent_volume_naming_enabled": image.launch_options.is_consistent_volume_naming_enabled if image.launch_options else None,
} if image.launch_options else None,
}
logger.info(f"Retrieved details for image {image_id}")
return image_details
except Exception as e:
logger.exception(f"Error getting image details: {e}")
raise
def list_shapes(compute_client: oci.core.ComputeClient, compartment_id: str) -> List[Dict[str, Any]]:
"""
List all shapes available in a compartment.
Args:
compute_client: OCI Compute client
compartment_id: OCID of the compartment
Returns:
List of shapes with their details
"""
try:
shapes_response = oci.pagination.list_call_get_all_results(
compute_client.list_shapes,
compartment_id
)
shapes = []
for shape in shapes_response.data:
shapes.append({
"shape": shape.shape,
"processor_description": shape.processor_description,
"ocpus": shape.ocpus,
"memory_in_gbs": shape.memory_in_gbs,
"networking_bandwidth_in_gbps": shape.networking_bandwidth_in_gbps,
"max_vnic_attachments": shape.max_vnic_attachments,
"gpus": shape.gpus,
"gpu_description": shape.gpu_description,
"local_disks": shape.local_disks,
"local_disks_total_size_in_gbs": shape.local_disks_total_size_in_gbs,
"local_disk_description": shape.local_disk_description,
"rdma_ports": shape.rdma_ports,
"rdma_bandwidth_in_gbps": shape.rdma_bandwidth_in_gbps,
"is_live_migration_supported": shape.is_live_migration_supported,
"ocpu_options": {
"min": shape.ocpu_options.min if shape.ocpu_options else None,
"max": shape.ocpu_options.max if shape.ocpu_options else None,
} if shape.ocpu_options else None,
"memory_options": {
"min_in_g_bs": shape.memory_options.min_in_g_bs if shape.memory_options else None,
"max_in_g_bs": shape.memory_options.max_in_g_bs if shape.memory_options else None,
"default_per_ocpu_in_g_bs": shape.memory_options.default_per_ocpu_in_g_bs if shape.memory_options else None,
"min_per_ocpu_in_g_bs": shape.memory_options.min_per_ocpu_in_g_bs if shape.memory_options else None,
"max_per_ocpu_in_g_bs": shape.memory_options.max_per_ocpu_in_g_bs if shape.memory_options else None,
} if shape.memory_options else None,
"networking_bandwidth_options": {
"min_in_gbps": shape.networking_bandwidth_options.min_in_gbps if shape.networking_bandwidth_options else None,
"max_in_gbps": shape.networking_bandwidth_options.max_in_gbps if shape.networking_bandwidth_options else None,
"default_per_ocpu_in_gbps": shape.networking_bandwidth_options.default_per_ocpu_in_gbps if shape.networking_bandwidth_options else None,
} if shape.networking_bandwidth_options else None,
})
logger.info(f"Found {len(shapes)} shapes in compartment {compartment_id}")
return shapes
except Exception as e:
logger.exception(f"Error listing shapes: {e}")
raise
def get_namespace(object_storage_client: oci.object_storage.ObjectStorageClient) -> Dict[str, Any]:
"""
Get the Object Storage namespace for the tenancy.
Args:
object_storage_client: OCI ObjectStorage client
Returns:
Object Storage namespace details
"""
try:
namespace = object_storage_client.get_namespace().data
namespace_details = {
"namespace": namespace,
}
logger.info(f"Retrieved Object Storage namespace: {namespace}")
return namespace_details
except Exception as e:
logger.exception(f"Error getting namespace: {e}")
raise
def list_regions(identity_client: oci.identity.IdentityClient) -> List[Dict[str, Any]]:
"""
List all available regions.
Args:
identity_client: OCI Identity client
Returns:
List of regions with their details
"""
try:
regions_response = identity_client.list_regions()
regions = []
for region in regions_response.data:
regions.append({
"key": region.key,
"name": region.name,
})
logger.info(f"Found {len(regions)} regions")
return regions
except Exception as e:
logger.exception(f"Error listing regions: {e}")
raise
def get_tenancy_info(identity_client: oci.identity.IdentityClient, tenancy_id: str) -> Dict[str, Any]:
"""
Get tenancy information.
Args:
identity_client: OCI Identity client
tenancy_id: OCID of the tenancy
Returns:
Tenancy details
"""
try:
tenancy = identity_client.get_tenancy(tenancy_id).data
tenancy_details = {
"id": tenancy.id,
"name": tenancy.name,
"description": tenancy.description,
"home_region_key": tenancy.home_region_key,
"upi_idcs_compatibility_layer_endpoint": tenancy.upi_idcs_compatibility_layer_endpoint,
}
logger.info(f"Retrieved tenancy details for {tenancy_id}")
return tenancy_details
except Exception as e:
logger.exception(f"Error getting tenancy details: {e}")
raise
