(self, data: dict, endpoint: str, timeout: int = TIMEOUT_SECS)
| 38 | self.headers['Authorization'] = f'Bearer {iap_token}' |
| 39 | |
| 40 | def make_post_request(self, data: dict, endpoint: str, timeout: int = TIMEOUT_SECS) -> dict: |
| 41 | url = f'{self.url_base}{endpoint}' |
| 42 | request_body = dict(data) |
| 43 | req = urllib.request.Request(url, headers=self.headers, data=bytes( |
| 44 | json.dumps(request_body), encoding="utf-8")) |
| 45 | try: |
| 46 | with urllib.request.urlopen(req, timeout=timeout) as response: |
| 47 | response_json = json.loads(response.read()) |
| 48 | if not response_json.get('wasSuccessful') or 'result' not in response_json: |
| 49 | raise BackendError( |
| 50 | f'Unexpected response from the service. Response: {response_json}.') |
| 51 | return response_json['result'] |
| 52 | except (URLError, socket.timeout) as e: |
| 53 | if isinstance( |
| 54 | e, socket.timeout) or isinstance( |
| 55 | e.reason, socket.timeout): |
| 56 | raise ConnectionError( |
| 57 | 'Timeout error trying to communicate with service. Please ensure internet is on.') from e |
| 58 | raise ConnectionError( |
| 59 | 'Connection error trying to communicate with service.') from e |
| 60 | except HTTPError as e: |
| 61 | if e.code == 401 or e.code == 403: |
| 62 | raise CredentialError( |
| 63 | f'Service responded with error code {e.code}.' |
| 64 | ' Please ensure you have access to the resource.') from e |
| 65 | raise BackendError('Unexpected response from the service.') from e |
no test coverage detected