def _refresh_credentials()

in alibabacloud_credentials/provider/uri.py [0:0]

• 32 lines of code
• 11 McCabe index (conditional complexity)

    def _refresh_credentials(self) -> RefreshResult[Credentials]:
        r = urlparse(self._uri)
        tea_request = ph.get_new_request()
        tea_request.headers['host'] = r.hostname
        tea_request.port = r.port
        tea_request.protocol = r.scheme or self._protocol or 'http'
        tea_request.method = 'GET'
        tea_request.pathname = r.path
        for key, values in parse_qs(r.query).items():
            for value in values:
                tea_request.query[key] = value

        response = TeaCore.do_action(tea_request, self._runtime_options)

        if response.status_code != 200:
            raise CredentialException(
                f'error refreshing credentials from {self._uri},  http_code={str(response.status_code)}, result: {response.body.decode("utf-8")}')

        body = response.body.decode('utf-8')

        dic = json.loads(body)
        content_code = dic.get('Code')

        if content_code != "Success" or 'AccessKeyId' not in dic or 'AccessKeySecret' not in dic or 'SecurityToken' not in dic or 'Expiration' not in dic:
            raise CredentialException(
                f'error retrieving credentials from {self._uri} result: {response.body.decode("utf-8")}')

        # 先转换为时间数组
        time_array = time.strptime(dic.get('Expiration'), '%Y-%m-%dT%H:%M:%SZ')
        # 转换为时间戳
        expiration = calendar.timegm(time_array)
        credentials = Credentials(
            access_key_id=dic.get('AccessKeyId'),
            access_key_secret=dic.get('AccessKeySecret'),
            security_token=dic.get('SecurityToken'),
            expiration=expiration,
            provider_name=self.get_provider_name()
        )
        return RefreshResult(value=credentials,
                             stale_time=_get_stale_time(expiration))