exceptions.py 1.6 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546
  1. import json
  2. from typing import Any, Dict, Optional
  3. from httpx import Headers, Response
  4. MAX_CONTENT = 200
  5. class ApiException(Exception):
  6. """Base class"""
  7. class UnexpectedResponse(ApiException):
  8. def __init__(self, status_code: Optional[int], reason_phrase: str, content: bytes, headers: Headers) -> None:
  9. self.status_code = status_code
  10. self.reason_phrase = reason_phrase
  11. self.content = content
  12. self.headers = headers
  13. @staticmethod
  14. def for_response(response: Response) -> "ApiException":
  15. return UnexpectedResponse(
  16. status_code=response.status_code,
  17. reason_phrase=response.reason_phrase,
  18. content=response.content,
  19. headers=response.headers,
  20. )
  21. def __str__(self) -> str:
  22. status_code_str = f"{self.status_code}" if self.status_code is not None else ""
  23. if self.reason_phrase == "" and self.status_code is not None:
  24. reason_phrase_str = "(Unrecognized Status Code)"
  25. else:
  26. reason_phrase_str = f"({self.reason_phrase})"
  27. status_str = f"{status_code_str} {reason_phrase_str}".strip()
  28. short_content = self.content if len(self.content) <= MAX_CONTENT else self.content[: MAX_CONTENT - 3] + b" ..."
  29. raw_content_str = f"Raw response content:\n{short_content!r}"
  30. return f"Unexpected Response: {status_str}\n{raw_content_str}"
  31. def structured(self) -> Dict[str, Any]:
  32. return json.loads(self.content)
  33. class ResponseHandlingException(ApiException):
  34. def __init__(self, source: Exception):
  35. self.source = source