Changeset View
Changeset View
Standalone View
Standalone View
swh/core/api/__init__.py
| Show First 20 Lines • Show All 115 Lines • ▼ Show 20 Lines | def __str__(self): | ||||
| return ( | return ( | ||||
| f"<RemoteException {self.response.status_code} " | f"<RemoteException {self.response.status_code} " | ||||
| f'{self.args[0]["type"]}: {self.args[0]["args"]}>' | f'{self.args[0]["type"]}: {self.args[0]["args"]}>' | ||||
| ) | ) | ||||
| else: | else: | ||||
| return super().__str__() | return super().__str__() | ||||
| class TransientRemoteException(RemoteException): | |||||
| """Subclass of RemoteException representing errors which are expected | |||||
| to be temporary. | |||||
| """ | |||||
| F = TypeVar("F", bound=Callable) | F = TypeVar("F", bound=Callable) | ||||
| def remote_api_endpoint(path: str, method: str = "POST") -> Callable[[F], F]: | def remote_api_endpoint(path: str, method: str = "POST") -> Callable[[F], F]: | ||||
| def dec(f: F) -> F: | def dec(f: F) -> F: | ||||
| f._endpoint_path = path # type: ignore | f._endpoint_path = path # type: ignore | ||||
| f._method = method # type: ignore | f._method = method # type: ignore | ||||
| return f | return f | ||||
| ▲ Show 20 Lines • Show All 197 Lines • ▼ Show 20 Lines | def raise_for_status(self, response) -> None: | ||||
| for exc_type in self.reraise_exceptions: | for exc_type in self.reraise_exceptions: | ||||
| if exc_type.__name__ == exc_data["type"]: | if exc_type.__name__ == exc_data["type"]: | ||||
| exception = exc_type(*exc_data["args"]) | exception = exc_type(*exc_data["args"]) | ||||
| break | break | ||||
| else: | else: | ||||
| exception = RemoteException(payload=exc_data, response=response) | exception = RemoteException(payload=exc_data, response=response) | ||||
| elif status_class == 5: | elif status_class == 5: | ||||
| cls: Type[RemoteException] | |||||
| if status_code == 503: | |||||
| # This isn't a generic HTTP client and we know the server does | |||||
| # not support the Retry-After header, so we do not implement | |||||
| # it here either. | |||||
| cls = TransientRemoteException | |||||
| else: | |||||
| cls = RemoteException | |||||
| exc_data = self._decode_response(response, check_status=False) | exc_data = self._decode_response(response, check_status=False) | ||||
| exception = RemoteException(payload=exc_data, response=response) | exception = cls(payload=exc_data, response=response) | ||||
| if exception: | if exception: | ||||
| raise exception from None | raise exception from None | ||||
| if status_class != 2: | if status_class != 2: | ||||
| raise RemoteException( | raise RemoteException( | ||||
| payload=f"API HTTP error: {status_code} {response.content}", | payload=f"API HTTP error: {status_code} {response.content}", | ||||
| response=response, | response=response, | ||||
| ▲ Show 20 Lines • Show All 154 Lines • Show Last 20 Lines | |||||