Add Sentry context provider

Links a failed task to its Sentry issue by reading back
sentry_sdk.last_event_id() rather than capturing the exception itself,
since the app is assumed to already report it via its own Sentry
integration. A snapshot taken when the task starts filters out stale
ids left over from an unrelated earlier capture.

Author: snopoke

pyproject.toml

@@ -47,6 +47,9 @@ cli = [
 procrastinate = [
     "procrastinate>=3.0",
 ]
+sentry = [
+    "sentry-sdk>=1.0",
+]
 
 [tool.uv]
 package = true
@@ -69,6 +72,7 @@ dev = [
     "redis",
     "openapi-python-client",
     "taskbadger[cli]",
+    "taskbadger[sentry]",
 ]
 
 [project.scripts]

taskbadger/context_providers/sentry.py

@@ -0,0 +1,43 @@
+from taskbadger.context_providers import ContextProvider
+
+
+class SentryContextProvider(ContextProvider):
+    """Links a failed task to the corresponding Sentry issue.
+
+    Reads back `sentry_sdk.last_event_id()` rather than capturing the exception
+    itself, on the assumption the surrounding system already reports its own
+    exceptions to Sentry (e.g. via a framework integration). To avoid linking to
+    a stale event left over from something unrelated, a snapshot is taken when
+    the task starts and the event id is only reported if it changed by the time
+    the task errors.
+
+    Requires the `sentry-sdk` package; a no-op if it isn't installed.
+    """
+
+    identifier = "sentry"
+
+    def __init__(self, organization_slug: str = None, base_url: str = "https://sentry.io"):
+        self.organization_slug = organization_slug
+        self.base_url = base_url.rstrip("/")
+
+    def snapshot(self):
+        try:
+            import sentry_sdk
+        except ImportError:
+            return None
+        return sentry_sdk.last_event_id()
+
+    def capture_error_context(self, exception: BaseException, snapshot=None) -> dict | None:
+        try:
+            import sentry_sdk
+        except ImportError:
+            return None
+
+        event_id = sentry_sdk.last_event_id()
+        if not event_id or event_id == snapshot:
+            return None
+
+        context = {"event_id": event_id}
+        if self.organization_slug:
+            context["url"] = f"{self.base_url}/organizations/{self.organization_slug}/issues/?query={event_id}"
+        return context

tests/test_context_providers_sentry.py

@@ -0,0 +1,52 @@
+import sys
+from unittest import mock
+
+from taskbadger.context_providers.sentry import SentryContextProvider
+
+
+def test_sentry_provider_not_installed():
+    provider = SentryContextProvider()
+    with mock.patch.dict(sys.modules, {"sentry_sdk": None}):
+        assert provider.snapshot() is None
+        assert provider.capture_error_context(ValueError("boom")) is None
+
+
+def test_sentry_provider_no_event_id():
+    provider = SentryContextProvider()
+    with mock.patch("sentry_sdk.last_event_id", return_value=None):
+        assert provider.capture_error_context(ValueError("boom")) is None
+
+
+def test_sentry_provider_stale_event_id_not_reported():
+    """If last_event_id() hasn't changed since the snapshot, the exception
+    was never actually captured by Sentry -- don't report the stale id."""
+    provider = SentryContextProvider()
+    with mock.patch("sentry_sdk.last_event_id", return_value="stale123"):
+        snapshot = provider.snapshot()
+        context = provider.capture_error_context(ValueError("boom"), snapshot)
+    assert context is None
+
+
+def test_sentry_provider_event_id_only():
+    provider = SentryContextProvider()
+    snapshot = None
+    with mock.patch("sentry_sdk.last_event_id", return_value="abc123"):
+        context = provider.capture_error_context(ValueError("boom"), snapshot)
+    assert context == {"event_id": "abc123"}
+
+
+def test_sentry_provider_with_url():
+    provider = SentryContextProvider(organization_slug="acme")
+    with mock.patch("sentry_sdk.last_event_id", return_value="abc123"):
+        context = provider.capture_error_context(ValueError("boom"))
+    assert context == {
+        "event_id": "abc123",
+        "url": "https://sentry.io/organizations/acme/issues/?query=abc123",
+    }
+
+
+def test_sentry_provider_custom_base_url():
+    provider = SentryContextProvider(organization_slug="acme", base_url="https://sentry.example.com/")
+    with mock.patch("sentry_sdk.last_event_id", return_value="abc123"):
+        context = provider.capture_error_context(ValueError("boom"))
+    assert context["url"] == "https://sentry.example.com/organizations/acme/issues/?query=abc123"