Add Starlette Request data builder

rollbar/__init__.py

@@ -89,6 +89,11 @@
 except ImportError:
  AppEngineFetch = None
 
+try:
+ from starlette.requests import Request as StarletteRequest
+except ImportError:
+ StarletteRequest = None
+
 
 def passthrough_decorator(func):
  def wrap(*args, **kwargs):
@@ -877,7 +882,8 @@ def _build_person_data(request):
  else:
  return None
 
- if hasattr(request, 'user'):
+ # TODO: Check if AuthenticationMiddlawre installed for Startlette
+ if not StarletteRequest and hasattr(request, 'user'):
  user_prop = request.user
  user = user_prop() if callable(user_prop) else user_prop
  if not user:
@@ -1120,6 +1126,10 @@ def _build_request_data(request):
  if isinstance(request, dict) and 'wsgi.version' in request:
  return _build_wsgi_request_data(request)
 
+ # Starlette
+ if StarletteRequest and isinstance(request, StarletteRequest):
+ return _build_starlette_request_data(request)
+
  return None
 
 
@@ -1299,6 +1309,16 @@ def _build_wsgi_request_data(request):
 
  return request_data
 
+def _build_starlette_request_data(request):
+ request_data = {
+ 'url': str(request.url),
+ 'GET': dict(request.query_params),
+ 'headers': dict(request.headers),
+ 'method': request.method,
+ }
+
+ return request_data
+
 
 def _filter_ip(request_data, capture_ip):
  if 'user_ip' not in request_data or capture_ip == True:

rollbar/contrib/starlette/__init__.py

@@ -1,17 +1,27 @@
+import sys
+
 from starlette import __version__
-from starlette.types import ASGIApp
+from starlette.requests import Request
+from starlette.types import Receive, Scope, Send
 
 import rollbar
 from rollbar.contrib.asgi import ASGIMiddleware
 
 
 class StarletteMiddleware(ASGIMiddleware):
- ...
+ async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+ try:
+ await self.app(scope, receive, send)
+ except Exception:
+ if scope["type"] == "http":
+ request = Request(scope, receive, send)
+ exc_info = sys.exc_info()
+ rollbar.report_exc_info(exc_info, request)
+ raise
 
 
 def _hook(request, data):
  data["framework"] = f"starlette {__version__}"
 
 
 rollbar.BASE_DATA_HOOK = _hook
-

rollbar/test/test_rollbar.py

@@ -117,6 +117,50 @@ def test_wsgi_request_data(self):
  self.assertDictEqual(data['GET'], {'format': 'json', 'param1': 'value1', 'param2': 'value2'})
  self.assertDictEqual(data['headers'], {'Connection': 'close', 'Host': 'example.com', 'User-Agent': 'Agent'})
 
+ def test_starlette_request_data(self):
+ try:
+ from starlette.requests import Request
+ except ImportError:
+ self.skipTest("Requires Starlette to be installed")
+
+ scope = {
+ "type": "http",
+ "client": ("127.0.0.1", 1453),
+ "headers": [
+ (b"accept", b"*/*"),
+ (b"content-type", b"application/x-www-form-urlencoded"),
+ (b"host", b"example.com"),
+ (b"user-agent", b"Agent"),
+ ],
+ "http_version": "1.1",
+ "method": "GET",
+ "path": "/api/test",
+ "query_params": {
+ "format": "json",
+ "param1": "value1",
+ "param2": "value2",
+ },
+ "query_string": b"format=json&param1=value1&param2=value2",
+ "scheme": "http",
+ "server": ("example.com", 80),
+ "url": {"path": "example.com", "root_path": "xxxx"},
+ }
+ request = Request(scope)
+ data = rollbar._build_starlette_request_data(request)
+
+ self.assertEqual(data["url"], "https://example.com/api/test?format=json&param1=value1&param2=value2")
+ self.assertEqual(data["method"], "GET")
+ self.assertDictEqual(data["GET"], {"format": "json", "param1": "value1", "param2": "value2"})
+ self.assertDictEqual(
+ data["headers"],
+ {
+ "accept": "*/*",
+ "content-type": "application/x-www-form-urlencoded",
+ "host": "example.com",
+ "user-agent": "Agent",
+ },
+ )
+
  @mock.patch('rollbar.send_payload')
  def test_report_exception(self, send_payload):
 

rollbar/test/test_starlette.py

@@ -65,6 +65,31 @@ def root(request):
  self.assertEqual(exc_type, ZeroDivisionError)
  self.assertIsInstance(exc_value, ZeroDivisionError)
 
+ def test_should_report_with_request_data(self):
+ from starlette.applications import Starlette
+ from starlette.middleware import Middleware
+ from starlette.requests import Request
+ from starlette.routing import Route
+ from starlette.testclient import TestClient
+ from rollbar.contrib.starlette import StarletteMiddleware
+
+ def root(request):
+ 1 / 0
+
+ routes = [ Route("/", root) ]
+ middleware = [ Middleware(StarletteMiddleware) ]
+ app = Starlette(routes=routes, middleware=middleware)
+
+ client = TestClient(app)
+ with mock.patch("rollbar.report_exc_info") as mock_report:
+ with self.assertRaises(ZeroDivisionError):
+ client.get("/")
+
+ mock_report.assert_called_once()
+ request = mock_report.call_args[0][1]
+
+ self.assertIsInstance(request, Request)
+
  def test_should_support_type_hints(self):
  from starlette.types import ASGIApp, Receive, Scope, Send