serverless.py 1.6 KB

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