in mds_plugin/bastion.py [0:0]
def get_bastion(**kwargs):
"""Gets a Bastion with the given id
If no id is given, it will prompt the user for the id.
Args:
**kwargs: Optional parameters
Keyword Args:
bastion_name (str): The new name bastion
bastion_id (str): OCID of the bastion.
await_state (str): Await the given lifecycle state, ACTIVE, DELETED, ..
ignore_current (bool): Whether the current bastion should be ignored
fallback_to_any_in_compartment (bool): Whether to lookup bastion in
compartment
compartment_id (str): OCID of the compartment.
config (dict): An OCI config object or None.
config_profile (str): The name of an OCI config profile
interactive (bool): Indicates whether to execute in interactive mode
return_type (str): "STR" will return a formatted string, "DICT" will
return the object converted to a dict structure and "OBJ" will
return the OCI Python object for internal plugin usage
raise_exceptions (bool): If set to true exceptions are raised
Returns:
None
"""
bastion_name = kwargs.get("bastion_name")
bastion_id = kwargs.get("bastion_id")
await_state = kwargs.get("await_state")
ignore_current = kwargs.get("ignore_current", False)
fallback_to_any_in_compartment = kwargs.get(
"fallback_to_any_in_compartment", False)
compartment_id = kwargs.get("compartment_id")
config = kwargs.get("config")
config_profile = kwargs.get("config_profile")
interactive = kwargs.get("interactive", core.get_interactive_default())
return_type = kwargs.get("return_type", core.RETURN_DICT)
raise_exceptions = kwargs.get(
"raise_exceptions", # On internal call (RETURN_OBJ), raise exceptions
True if return_type == core.RETURN_OBJ else not interactive)
# Get the active config and compartment
try:
config = configuration.get_current_config(
config=config, config_profile=config_profile,
interactive=interactive)
if not ignore_current and bastion_name is None:
bastion_id = configuration.get_current_bastion_id(
bastion_id=bastion_id, config=config)
import oci.exceptions
try:
# Initialize the Bastion client
bastion_client = core.get_oci_bastion_client(config=config)
if not interactive and not bastion_id and not bastion_name:
raise ValueError('No bastion_id nor bastion_name given.')
bastion = None
if bastion_id:
bastion = bastion_client.get_bastion(
bastion_id=bastion_id).data
if not bastion:
raise ValueError('The bastion with the given OCID '
f'{bastion_id} was not found.')
if not bastion and (bastion_name or interactive):
# List the Bastion of the current compartment
bastions = bastion_client.list_bastions(
compartment_id=compartment_id).data
# Filter out all deleted compartments
bastions = [
u for u in bastions if u.lifecycle_state != "DELETED"]
if len(bastions) == 0:
if interactive:
print("No Bastions available in this compartment.")
return
# If a bastion name was given, look it up
if bastion_name:
for b in bastions:
if b.name == bastion_name:
bastion = bastion_client.get_bastion(
bastion_id=b.id).data
break
if bastion is None and not interactive:
raise ValueError(f"Bastion {bastion_name} was not "
"found in this compartment.")
# Fallback to first in compartment
if fallback_to_any_in_compartment:
bastion = bastion_client.get_bastion(
bastion_id=bastions[0].id).data
if not bastion and interactive:
# If the user_name was not given or found, print out the list
bastion_list = format_bastion_listing(items=bastions)
if bastion_name is None:
print(bastion_list)
# Let the user choose from the list
selected_bastion = core.prompt_for_list_item(
item_list=bastions,
prompt_caption=("Please enter the name or index "
"of the Bastion: "),
item_name_property="name", given_value=bastion_name)
if selected_bastion:
bastion = bastion_client.get_bastion(
bastion_id=selected_bastion.id).data
if bastion and await_state:
import time
if interactive:
print(f'Waiting for Bastion to reach '
f'{await_state} state...')
bastion_id = bastion.id
# Wait for the Bastion Session to reach state await_state
cycles = 0
while cycles < 48:
bastion = bastion_client.get_bastion(
bastion_id=bastion_id).data
if bastion.lifecycle_state == await_state:
break
else:
time.sleep(5)
s = "." * (cycles + 1)
if interactive:
print(f'Waiting for Bastion to reach '
f'{await_state} state...{s}')
cycles += 1
if bastion.lifecycle_state != await_state:
raise Exception("Bastion did not reach the state "
f"{await_state} within 4 minutes.")
return core.oci_object(
oci_object=bastion,
return_type=return_type,
format_function=format_bastion_listing)
except oci.exceptions.ServiceError as e:
if interactive:
raise
print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
return
except Exception as e:
if raise_exceptions:
raise
print(f'ERROR: {e}')