in mds_plugin/object_store.py [0:0]
def delete_bucket_object(name=None, **kwargs):
"""Deletes an object store bucket objects
Args:
name (str): The name of the object, can include * to match multiple
objects
**kwargs: Additional options
Keyword Args:
bucket_name (str): The name of the bucket.
compartment_id (str): OCID of the parent compartment.
config (object): An OCI config object or None.
config_profile (str): The name of an OCI config profile
interactive (bool): If set to false, function returns true on success
Returns:
None
"""
compartment_id = kwargs.get("compartment_id")
config = kwargs.get("config")
config_profile = kwargs.get("config_profile")
bucket_name = kwargs.get("bucket_name")
interactive = kwargs.get("interactive", core.get_interactive_default())
raise_exceptions = kwargs.get("raise_exceptions", not interactive)
# Get the active config and compartment
try:
# Get the active config and compartment
config = configuration.get_current_config(
config=config, config_profile=config_profile,
interactive=interactive)
compartment_id = configuration.get_current_compartment_id(
compartment_id=compartment_id, config=config)
bucket_name = configuration.get_current_bucket_name(
bucket_name=bucket_name, config=config)
import oci.object_storage
import mysqlsh
import re
bucket = get_bucket(
bucket_name=bucket_name, compartment_id=compartment_id,
config=config)
if bucket is None:
if interactive:
print("Operation Cancelled.\n")
return
# Initialize the Object Store client
os_client = core.get_oci_object_storage_client(config=config)
# Get Object Store namespace
namespace_name = get_object_store_namespace(config)
# If the user specified * as name, delete all
if name and (name == '*' or '*' in name):
# Get object list
objects = oci.pagination.list_call_get_all_results(
os_client.list_objects,
namespace_name=namespace_name,
bucket_name=bucket.name,
limit=1000).data.objects
# Filter list
if name != '*':
name = name.lower()
# Filter list if PARs
if '*' in name:
name_pattern = '^' + name.replace('*', '.+')
objects = [obj for obj in objects
if re.search(name_pattern, obj.name.lower())]
else:
objects = [obj for obj in objects
if name == obj.name.lower()]
# Get object count
obj_count = len(objects)
if obj_count == 0:
if interactive:
print("No matching objects found for deletion.")
return
else:
raise ValueError("No matching objects found for deletion.")
# Prompt the user for confirmation
if interactive:
prompt = mysqlsh.globals.shell.prompt(
f"Are you sure you want to delete {obj_count} object"
f"{'s' if obj_count > 1 else ''} from {bucket.name} "
f"[yes/NO]: ",
{'defaultValue': 'no'}).strip().lower()
if prompt != "yes":
print("Deletion aborted.\n")
return
# Delete all objects
print(f"Deleting {obj_count} "
f"object{'s' if obj_count > 1 else ''}.")
import threading
thread_count = NTHREAD if obj_count > NTHREAD else obj_count
ths = [threading.Thread(
target=delete_file_from_list_from_bucket,
args=(i, os_client, objects, namespace_name, bucket.name,
thread_count))
for i in range(thread_count)]
for th in ths:
th.daemon = True
th.start()
for th in ths:
th.join()
if interactive:
print(f"Bucket object{'s' if '*' in name else ''} "
f"deleted successfully.")
elif name:
os_client.delete_object(
namespace_name=namespace_name, bucket_name=bucket.name,
object_name=name)
if interactive:
print(f"Bucket object '{name}' deleted successfully.")
elif interactive:
# Get object list
bucket_objects = oci.pagination.list_call_get_all_results(
os_client.list_objects,
namespace_name=namespace_name,
bucket_name=bucket.name,
limit=1000).data.objects
print(format_bucket_objects_listing(bucket_objects=bucket_objects))
obj_summary = core.prompt_for_list_item(
item_list=bucket_objects,
prompt_caption="Please enter the index or name of an object: ",
item_name_property="name")
if obj_summary is None:
print("Operation cancelled.")
return
name = obj_summary.name
os_client.delete_object(
namespace_name=namespace_name, bucket_name=bucket.name,
object_name=name)
print(f"Bucket object '{name}' deleted successfully.")
else:
if interactive:
print('No object name given.')
return
raise ValueError("No object name given.")
except oci.exceptions.ServiceError as e:
if raise_exceptions:
raise
print(f'ERROR: {e.message}. (Code: {e.code}; Status: {e.status})')
except Exception as e:
if raise_exceptions:
raise
print(f'Could not create the bucket objects.\n'
f'ERROR: {str(e)}')