feat: Add serverless integration

Author: untitaker

sentry_sdk/integrations/serverless.py

@@ -0,0 +1,46 @@
+import functools
+import sys
+
+from sentry_sdk import Hub
+from sentry_sdk.utils import event_from_exception
+from sentry_sdk._compat import reraise
+
+
+def serverless_function(f=None, flush=True):
+    def wrapper(f):
+        @functools.wraps(f)
+        def inner(*args, **kwargs):
+            try:
+                return f(*args, **kwargs)
+            except Exception:
+                _capture_and_reraise()
+            finally:
+                if flush:
+                    _flush_client()
+
+        return inner
+
+    if f is None:
+        return wrapper
+    else:
+        return wrapper(f)
+
+
+def _capture_and_reraise():
+    exc_info = sys.exc_info()
+    hub = Hub.current
+    if hub is not None and 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():
+    hub = Hub.current
+    if hub is not None:
+        hub.flush()

tests/integrations/serverless/test_serverless.py

@@ -0,0 +1,45 @@
+import pytest
+
+from sentry_sdk.integrations.serverless import serverless_function
+
+
+def test_basic(sentry_init, capture_exceptions, monkeypatch):
+    sentry_init()
+    exceptions = capture_exceptions()
+
+    flush_calls = []
+
+    monkeypatch.setattr("sentry_sdk.Hub.current.flush", lambda: flush_calls.append(1))
+
+    @serverless_function
+    def foo():
+        1 / 0
+
+    with pytest.raises(ZeroDivisionError):
+        foo()
+
+    exception, = exceptions
+    assert isinstance(exception, ZeroDivisionError)
+
+    assert flush_calls == [1]
+
+
+def test_flush_disabled(sentry_init, capture_exceptions, monkeypatch):
+    sentry_init()
+    exceptions = capture_exceptions()
+
+    flush_calls = []
+
+    monkeypatch.setattr("sentry_sdk.Hub.current.flush", lambda: flush_calls.append(1))
+
+    @serverless_function(flush=False)
+    def foo():
+        1 / 0
+
+    with pytest.raises(ZeroDivisionError):
+        foo()
+
+    exception, = exceptions
+    assert isinstance(exception, ZeroDivisionError)
+
+    assert flush_calls == []