(self, request)
| 56 | self.target = target |
| 57 | |
| 58 | def refresh(self, request): |
| 59 | try: |
| 60 | client = UserSecretsClient() |
| 61 | if self.target == GcpTarget.BIGQUERY: |
| 62 | self.token, self.expiry = client.get_bigquery_access_token() |
| 63 | elif self.target == GcpTarget.GCS: |
| 64 | self.token, self.expiry = client._get_gcs_access_token() |
| 65 | elif self.target == GcpTarget.CLOUDAI: |
| 66 | self.token, self.expiry = client._get_cloudai_access_token() |
| 67 | except ConnectionError as e: |
| 68 | logging.error(f"Connection error trying to refresh access token: {e}") |
| 69 | print("There was a connection error trying to fetch the access token. " |
| 70 | f"Please ensure internet is on in order to use the {self.target.service} Integration.") |
| 71 | raise RefreshError('Unable to refresh access token due to connection error.') from e |
| 72 | except Exception as e: |
| 73 | logging.error(f"Error trying to refresh access token: {e}") |
| 74 | if (not get_integrations().has_integration(self.target)): |
| 75 | logging.error(f"No {self.target.service} integration found.") |
| 76 | print( |
| 77 | f"Please ensure you have selected a {self.target.service} account in the Notebook Add-ons menu.") |
| 78 | raise RefreshError('Unable to refresh access token.') from e |
| 79 | |
| 80 | class KaggleKernelWithProjetCredentials(KaggleKernelCredentials): |
| 81 | """ Wrapper Kaggle Credentials with quota_project_id. |
nothing calls this directly
no test coverage detected