def find_and_clean()

in tools/ocivaultclean.py [0:0]

• 40 lines of code
• 10 McCabe index (conditional complexity)

def find_and_clean(profile, compartment_id, vault_id, ignore_list, confirm):
    config = config_from_file(profile_name=profile)
    vault_client = VaultsClient(config)
    secrets = list_call_get_all_results_generator(
        vault_client.list_secrets,
        yield_mode="record",
        compartment_id=compartment_id)

    secrets = filter(lambda item: item.vault_id == vault_id, secrets)
    secrets = filter(lambda item: item.lifecycle_state == "ACTIVE", secrets)
    secrets = filter(lambda item: "no-delete" not in item.freeform_tags, secrets)
    secrets = filter(lambda item: item.id not in ignore_list, secrets)

    try:
        # As we are using a generator only here we will do a network call to
        # OCI, thus auth errors etc. are reported from here.
        # We only *have to* make this a list in confirm mode, but this makes
        # things simpler
        todelete = list(secrets)
    except ServiceError as exc:
        eprint("\033[0;31mFailed to load Secrets from OCI. Error:")
        eprint(f"\033[0;32m{exc.message}\033[0m")
        sys.exit(1)

    if not todelete:
        if confirm:
            print("\033[0;31mNothing to do!\033[0m")
        return

    if confirm:
        print("\033[0;31mThese Screts are being marked for deletion in 24h:\033[0m")
        print("\n".join(map(lambda item: f"\033[0;32m{item.secret_name}\n\033[0;33m{item.id}\033[0m", todelete)))
        try:
            input(f'\033[0;31mPress Enter to delete {len(todelete)} Secrets or Ctrl-C to abort\033[0m')
        except KeyboardInterrupt:
            return

        counter = 0

    for item in todelete:
        # using one day and one minute, so that it is guaranteed more than minimal time, also recaclulating
        # in each iteration in case the loop runs longer than a minute
        delete_time = datetime.utcnow() + timedelta(days=1, minutes=1)

        if confirm:
            counter = counter + 1
            print(f'\033[0;33m({counter}/{len(todelete)}) \033[0;32m{item.secret_name :{" "}<{70}.70}\033[0m', end='\r')

        vault_client.schedule_secret_deletion(
                item.id,
                ScheduleSecretDeletionDetails(time_of_deletion=delete_time.strftime('%Y-%m-%dT%TZ'))
                )

    if confirm:
        # Make sure the last update line can be read
        print()