FilesExpand file tree

master

/

serverless.py

Copy path

Blame

More file actions

Blame

More file actions

Latest commit

 

History

History

History

85 lines (63 loc) · 1.92 KB

master

/

serverless.py

Copy path

Top

File metadata and controls

• Code
• Blame

85 lines (63 loc) · 1.92 KB

Raw

Copy raw file

Download raw file

Open symbols panel

Edit and raw actions

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

78

79

80

81

82

83

84

85

import sys

from sentry_sdk.hub import Hub

from sentry_sdk.utils import event_from_exception

from sentry_sdk._compat import reraise

from sentry_sdk._functools import wraps

from sentry_sdk._types import MYPY

if MYPY:

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): # noqa: F811

# 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 Hub(Hub.current) as hub:

with hub.configure_scope() as scope:

scope.clear_breadcrumbs()

try:

return f(*args, **kwargs)

except Exception:

_capture_and_reraise()

finally:

if flush:

_flush_client()

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()

hub = Hub.current

if hub.client is not None:

event, hint = event_from_exception(

exc_info,

client_options=hub.client.options,

mechanism={"type": "serverless", "handled": False},

)

hub.capture_event(event, hint=hint)

reraise(*exc_info)

def _flush_client():

# type: () -> None

return Hub.current.flush()