def _begin_delete_secret()

in azext_edge/edge/providers/orchestration/resources/connector/opcua/certs.py [0:0]

• 23 lines of code
• 7 McCabe index (conditional complexity)

    def _begin_delete_secret(self, keyvault_name: str, secret_name: str):
        # Construct vault URL
        vault_url = KEYVAULT_URL.format(keyvaultName=keyvault_name)

        # Initiate deletion
        pipeline_response = self.keyvault_client.delete_secret(
            vault_base_url=vault_url,
            secret_name=secret_name,
            cls=lambda pipeline_response, _, __: pipeline_response,
        )

        for attempt in range(SECRET_DELETE_MAX_RETRIES):
            try:
                # Check if secret is deleted
                self.keyvault_client.get_deleted_secret(
                    vault_base_url=vault_url,
                    secret_name=secret_name,
                )
                return  # Exit if no exception, deletion confirmed
            except ResourceNotFoundError:
                # Secret not yet deleted; retry after delay
                transport: HttpTransport = cast(HttpTransport, pipeline_response.context.transport)
                transport.sleep(SECRET_DELETE_RETRY_INTERVAL)
                attempt += 1
            except HttpResponseError as e:
                if e.status_code == 403:
                    # Permission issue encountered; exit loop
                    break
                raise

        # Failed to confirm deletion after retries
        raise TimeoutError(f"Failed to delete secret '{secret_name}' within {SECRET_DELETE_MAX_RETRIES} retries.")