serverless.py 1.8 KB

12345678910111213141516171819202122232425262728293031323334353637383940414243444546474849505152535455565758596061626364656667686970717273747576
  1. import sys
  2. from functools import wraps
  3. import sentry_sdk
  4. from sentry_sdk.utils import event_from_exception, reraise
  5. from typing import TYPE_CHECKING
  6. if TYPE_CHECKING:
  7. from typing import Any
  8. from typing import Callable
  9. from typing import TypeVar
  10. from typing import Union
  11. from typing import Optional
  12. from typing import overload
  13. F = TypeVar("F", bound=Callable[..., Any])
  14. else:
  15. def overload(x):
  16. # type: (F) -> F
  17. return x
  18. @overload
  19. def serverless_function(f, flush=True):
  20. # type: (F, bool) -> F
  21. pass
  22. @overload
  23. def serverless_function(f=None, flush=True): # noqa: F811
  24. # type: (None, bool) -> Callable[[F], F]
  25. pass
  26. def serverless_function(f=None, flush=True): # noqa
  27. # type: (Optional[F], bool) -> Union[F, Callable[[F], F]]
  28. def wrapper(f):
  29. # type: (F) -> F
  30. @wraps(f)
  31. def inner(*args, **kwargs):
  32. # type: (*Any, **Any) -> Any
  33. with sentry_sdk.isolation_scope() as scope:
  34. scope.clear_breadcrumbs()
  35. try:
  36. return f(*args, **kwargs)
  37. except Exception:
  38. _capture_and_reraise()
  39. finally:
  40. if flush:
  41. sentry_sdk.flush()
  42. return inner # type: ignore
  43. if f is None:
  44. return wrapper
  45. else:
  46. return wrapper(f)
  47. def _capture_and_reraise():
  48. # type: () -> None
  49. exc_info = sys.exc_info()
  50. client = sentry_sdk.get_client()
  51. if client.is_active():
  52. event, hint = event_from_exception(
  53. exc_info,
  54. client_options=client.options,
  55. mechanism={"type": "serverless", "handled": False},
  56. )
  57. sentry_sdk.capture_event(event, hint=hint)
  58. reraise(*exc_info)