Class UserSecretsClient

patches/kaggle_secrets.py:43–150

class UserSecretsClient():
    GET_USER_SECRET_ENDPOINT = '/requests/GetUserSecretRequest'
    GET_USER_SECRET_BY_LABEL_ENDPOINT = '/requests/GetUserSecretByLabelRequest'

    def __init__(self):
        self.web_client = KaggleWebClient()

    def get_secret(self, label) -> str:
        """Retrieves a user secret value by its label.

        This returns the value of the secret with the given label,
        if it attached to the current kernel.
        Example usage:
            client = UserSecretsClient()
            secret = client.get_secret('my_db_password')
        """
        if label is None or len(label) == 0:
            raise ValidationError("Label must be non-empty.")
        request_body = {
            'Label': label,
        }
        response_json = self.web_client.make_post_request(request_body, self.GET_USER_SECRET_BY_LABEL_ENDPOINT)
        if 'secret' not in response_json:
            raise BackendError(
                f'Unexpected response from the service. Response: {response_json}')
        return response_json['secret']

    def get_gcloud_credential(self) -> str:
        """Retrieves the Google Cloud SDK credential attached to the current
        kernel.
        Example usage:
            client = UserSecretsClient()
            credential_json = client.get_gcloud_credential()
        """
        try:
            return self.get_secret("__gcloud_sdk_auth__")
        except BackendError as backend_error:
            message = str(backend_error.args)
            if message.find('No user secrets exist') != -1:
              raise NotFoundError('Google Cloud SDK credential not found.')
            else:
              raise

    def set_gcloud_credentials(self, project=None, account=None):
        """Set user credentials attached to the current kernel and optionally the project & account name to the `gcloud` CLI.

        Example usage:
            client = UserSecretsClient()
            client.set_gcloud_credentials(project="my-gcp-project", account="me@my-org.com")

            !gcloud ai-platform jobs list
        """
        creds = self.get_gcloud_credential()
        creds_path = self._write_credentials_file(creds)

        subprocess.run(['gcloud', 'config', 'set', 'auth/credential_file_override', creds_path])

        if project: