"""
Tools for managing OCI Database resources.
"""
import logging
from typing import Dict, List, Any, Optional
import oci
logger = logging.getLogger(__name__)
def list_db_systems(database_client: oci.database.DatabaseClient, compartment_id: str) -> List[Dict[str, Any]]:
"""
List all DB systems in a compartment.
Args:
database_client: OCI Database client
compartment_id: OCID of the compartment
Returns:
List of DB systems with their details
"""
try:
db_systems_response = oci.pagination.list_call_get_all_results(
database_client.list_db_systems,
compartment_id
)
db_systems = []
for db_system in db_systems_response.data:
db_systems.append({
"id": db_system.id,
"display_name": db_system.display_name,
"compartment_id": db_system.compartment_id,
"availability_domain": db_system.availability_domain,
"shape": db_system.shape,
"cpu_core_count": db_system.cpu_core_count,
"node_count": db_system.node_count,
"database_edition": db_system.database_edition,
"lifecycle_state": db_system.lifecycle_state,
"time_created": str(db_system.time_created),
"data_storage_size_in_gbs": db_system.data_storage_size_in_gbs,
"data_storage_percentage": db_system.data_storage_percentage,
"license_model": db_system.license_model,
"version": db_system.version,
"hostname": db_system.hostname,
"domain": db_system.domain,
"backup_subnet_id": db_system.backup_subnet_id,
"subnet_id": db_system.subnet_id,
})
logger.info(f"Found {len(db_systems)} DB systems in compartment {compartment_id}")
return db_systems
except Exception as e:
logger.exception(f"Error listing DB systems: {e}")
raise
def get_db_system(database_client: oci.database.DatabaseClient, db_system_id: str) -> Dict[str, Any]:
"""
Get details of a specific DB system.
Args:
database_client: OCI Database client
db_system_id: OCID of the DB system
Returns:
Details of the DB system
"""
try:
db_system = database_client.get_db_system(db_system_id).data
db_system_details = {
"id": db_system.id,
"display_name": db_system.display_name,
"compartment_id": db_system.compartment_id,
"availability_domain": db_system.availability_domain,
"shape": db_system.shape,
"cpu_core_count": db_system.cpu_core_count,
"node_count": db_system.node_count,
"database_edition": db_system.database_edition,
"lifecycle_state": db_system.lifecycle_state,
"time_created": str(db_system.time_created),
"data_storage_size_in_gbs": db_system.data_storage_size_in_gbs,
"data_storage_percentage": db_system.data_storage_percentage,
"license_model": db_system.license_model,
"version": db_system.version,
"hostname": db_system.hostname,
"domain": db_system.domain,
"backup_subnet_id": db_system.backup_subnet_id,
"subnet_id": db_system.subnet_id,
"cluster_name": db_system.cluster_name,
"maintenance_window": {
"preference": db_system.maintenance_window.preference if db_system.maintenance_window else None,
"days_of_week": [day.name for day in db_system.maintenance_window.days_of_week] if db_system.maintenance_window and db_system.maintenance_window.days_of_week else None,
} if db_system.maintenance_window else None,
}
logger.info(f"Retrieved details for DB system {db_system_id}")
return db_system_details
except Exception as e:
logger.exception(f"Error getting DB system details: {e}")
raise
def list_databases(database_client: oci.database.DatabaseClient, compartment_id: str,
db_system_id: Optional[str] = None) -> List[Dict[str, Any]]:
"""
List all databases in a compartment, optionally filtered by DB system.
Args:
database_client: OCI Database client
compartment_id: OCID of the compartment
db_system_id: Optional OCID of the DB system to filter by
Returns:
List of databases with their details
"""
try:
databases_response = oci.pagination.list_call_get_all_results(
database_client.list_databases,
compartment_id,
db_system_id=db_system_id
)
databases = []
for database in databases_response.data:
databases.append({
"id": database.id,
"db_name": database.db_name,
"compartment_id": database.compartment_id,
"character_set": database.character_set,
"ncharacter_set": database.ncharacter_set,
"db_workload": database.db_workload,
"pdb_name": database.pdb_name,
"lifecycle_state": database.lifecycle_state,
"time_created": str(database.time_created),
"db_unique_name": database.db_unique_name,
"db_system_id": database.db_system_id,
"vm_cluster_id": database.vm_cluster_id,
"kms_key_id": database.kms_key_id,
"vault_id": database.vault_id,
})
logger.info(f"Found {len(databases)} databases in compartment {compartment_id}")
return databases
except Exception as e:
logger.exception(f"Error listing databases: {e}")
raise
def get_database(database_client: oci.database.DatabaseClient, database_id: str) -> Dict[str, Any]:
"""
Get details of a specific database.
Args:
database_client: OCI Database client
database_id: OCID of the database
Returns:
Details of the database
"""
try:
database = database_client.get_database(database_id).data
database_details = {
"id": database.id,
"db_name": database.db_name,
"compartment_id": database.compartment_id,
"character_set": database.character_set,
"ncharacter_set": database.ncharacter_set,
"db_workload": database.db_workload,
"pdb_name": database.pdb_name,
"lifecycle_state": database.lifecycle_state,
"time_created": str(database.time_created),
"db_unique_name": database.db_unique_name,
"db_system_id": database.db_system_id,
"vm_cluster_id": database.vm_cluster_id,
"kms_key_id": database.kms_key_id,
"vault_id": database.vault_id,
"source_database_point_in_time_recovery_timestamp": str(database.source_database_point_in_time_recovery_timestamp) if database.source_database_point_in_time_recovery_timestamp else None,
}
logger.info(f"Retrieved details for database {database_id}")
return database_details
except Exception as e:
logger.exception(f"Error getting database details: {e}")
raise
def list_autonomous_databases(database_client: oci.database.DatabaseClient, compartment_id: str) -> List[Dict[str, Any]]:
"""
List all autonomous databases in a compartment.
Args:
database_client: OCI Database client
compartment_id: OCID of the compartment
Returns:
List of autonomous databases with their details
"""
try:
autonomous_databases_response = oci.pagination.list_call_get_all_results(
database_client.list_autonomous_databases,
compartment_id
)
autonomous_databases = []
for adb in autonomous_databases_response.data:
autonomous_databases.append({
"id": adb.id,
"db_name": adb.db_name,
"display_name": adb.display_name,
"compartment_id": adb.compartment_id,
"lifecycle_state": adb.lifecycle_state,
"time_created": str(adb.time_created),
"cpu_core_count": adb.cpu_core_count,
"data_storage_size_in_tbs": adb.data_storage_size_in_tbs,
"is_free_tier": adb.is_free_tier,
"is_auto_scaling_enabled": adb.is_auto_scaling_enabled,
"db_workload": adb.db_workload,
"db_version": adb.db_version,
"license_model": adb.license_model,
"is_dedicated": adb.is_dedicated,
"autonomous_container_database_id": adb.autonomous_container_database_id,
"is_access_control_enabled": adb.is_access_control_enabled,
"whitelisted_ips": adb.whitelisted_ips,
"are_primary_whitelisted_ips_used": adb.are_primary_whitelisted_ips_used,
"standby_whitelisted_ips": adb.standby_whitelisted_ips,
"is_data_guard_enabled": adb.is_data_guard_enabled,
"is_local_data_guard_enabled": adb.is_local_data_guard_enabled,
"subnet_id": adb.subnet_id,
"nsg_ids": adb.nsg_ids,
"private_endpoint": adb.private_endpoint,
"private_endpoint_label": adb.private_endpoint_label,
})
logger.info(f"Found {len(autonomous_databases)} autonomous databases in compartment {compartment_id}")
return autonomous_databases
except Exception as e:
logger.exception(f"Error listing autonomous databases: {e}")
raise
def get_autonomous_database(database_client: oci.database.DatabaseClient, autonomous_database_id: str) -> Dict[str, Any]:
"""
Get details of a specific autonomous database.
Args:
database_client: OCI Database client
autonomous_database_id: OCID of the autonomous database
Returns:
Details of the autonomous database
"""
try:
adb = database_client.get_autonomous_database(autonomous_database_id).data
adb_details = {
"id": adb.id,
"db_name": adb.db_name,
"display_name": adb.display_name,
"compartment_id": adb.compartment_id,
"lifecycle_state": adb.lifecycle_state,
"time_created": str(adb.time_created),
"cpu_core_count": adb.cpu_core_count,
"data_storage_size_in_tbs": adb.data_storage_size_in_tbs,
"is_free_tier": adb.is_free_tier,
"is_auto_scaling_enabled": adb.is_auto_scaling_enabled,
"db_workload": adb.db_workload,
"db_version": adb.db_version,
"license_model": adb.license_model,
"is_dedicated": adb.is_dedicated,
"autonomous_container_database_id": adb.autonomous_container_database_id,
"is_access_control_enabled": adb.is_access_control_enabled,
"whitelisted_ips": adb.whitelisted_ips,
"are_primary_whitelisted_ips_used": adb.are_primary_whitelisted_ips_used,
"standby_whitelisted_ips": adb.standby_whitelisted_ips,
"is_data_guard_enabled": adb.is_data_guard_enabled,
"is_local_data_guard_enabled": adb.is_local_data_guard_enabled,
"subnet_id": adb.subnet_id,
"nsg_ids": adb.nsg_ids,
"private_endpoint": adb.private_endpoint,
"private_endpoint_label": adb.private_endpoint_label,
"connection_strings": {
"high": adb.connection_strings.high if adb.connection_strings else None,
"medium": adb.connection_strings.medium if adb.connection_strings else None,
"low": adb.connection_strings.low if adb.connection_strings else None,
"dedicated": adb.connection_strings.dedicated if adb.connection_strings else None,
} if adb.connection_strings else None,
"connection_urls": {
"sql_dev_web_url": adb.connection_urls.sql_dev_web_url if adb.connection_urls else None,
"apex_url": adb.connection_urls.apex_url if adb.connection_urls else None,
"machine_learning_user_management_url": adb.connection_urls.machine_learning_user_management_url if adb.connection_urls else None,
"graph_studio_url": adb.connection_urls.graph_studio_url if adb.connection_urls else None,
"mongo_db_url": adb.connection_urls.mongo_db_url if adb.connection_urls else None,
} if adb.connection_urls else None,
}
logger.info(f"Retrieved details for autonomous database {autonomous_database_id}")
return adb_details
except Exception as e:
logger.exception(f"Error getting autonomous database details: {e}")
raise
