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