base.py 2.7KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576777879808182
  1. # -*- coding: utf-8 -*-
  2. import json
  3. import typing
  4. try: # Python 3.5+
  5. from http import HTTPStatus
  6. except ImportError:
  7. from http import client as HTTPStatus
  8. from multidict import MultiDict
  9. import bottle
  10. from hapic.ext.bottle import BottleContext
  11. from hapic.processor import RequestParameters
  12. from hapic.processor import ProcessValidationError
  13. class Base(object):
  14. pass
  15. # TODO BS 20171105: Make this bottle agnostic !
  16. class MyContext(BottleContext):
  17. def __init__(
  18. self,
  19. app,
  20. fake_path_parameters=None,
  21. fake_query_parameters=None,
  22. fake_body_parameters=None,
  23. fake_form_parameters=None,
  24. fake_header_parameters=None,
  25. fake_files_parameters=None,
  26. ) -> None:
  27. super().__init__(app=app)
  28. self._handled_exceptions = [] # type: typing.List[typing.Tuple[typing.Type[Exception], int]] # nopep8
  29. self._exceptions_handler_installed = False
  30. self.fake_path_parameters = fake_path_parameters or {}
  31. self.fake_query_parameters = fake_query_parameters or MultiDict()
  32. self.fake_body_parameters = fake_body_parameters or {}
  33. self.fake_form_parameters = fake_form_parameters or MultiDict()
  34. self.fake_header_parameters = fake_header_parameters or {}
  35. self.fake_files_parameters = fake_files_parameters or {}
  36. def get_request_parameters(self, *args, **kwargs) -> RequestParameters:
  37. return RequestParameters(
  38. path_parameters=self.fake_path_parameters,
  39. query_parameters=self.fake_query_parameters,
  40. body_parameters=self.fake_body_parameters,
  41. form_parameters=self.fake_form_parameters,
  42. header_parameters=self.fake_header_parameters,
  43. files_parameters=self.fake_files_parameters,
  44. )
  45. def get_validation_error_response(
  46. self,
  47. error: ProcessValidationError,
  48. http_code: HTTPStatus=HTTPStatus.BAD_REQUEST,
  49. ) -> typing.Any:
  50. return self.get_response(
  51. response=json.dumps({
  52. 'original_error': {
  53. 'details': error.details,
  54. 'message': error.message,
  55. },
  56. 'http_code': http_code,
  57. }),
  58. http_code=http_code,
  59. )
  60. def _add_exception_class_to_catch(
  61. self,
  62. exception_class: typing.Type[Exception],
  63. http_code: int,
  64. ) -> None:
  65. if not self._exceptions_handler_installed:
  66. self._install_exceptions_handler()
  67. self._handled_exceptions.append((exception_class, http_code))
  68. def _get_handled_exception_classes(
  69. self,
  70. ) -> typing.List[typing.Tuple[typing.Type[Exception], int]]:
  71. return self._handled_exceptions