feat: filter validation error from sentry. centralize config

src/deriver/queue_manager.py

@@ -28,6 +28,7 @@ from src.dreamer.dream_scheduler import (
     set_dream_scheduler,
 )
 from src.models import QueueItem
+from src.sentry import initialize_sentry
 from src.utils.work_unit import parse_work_unit_key
 from src.webhooks.events import (
     QueueEmptyEvent,
@@ -68,15 +69,7 @@ class QueueManager:
 
         # Initialize Sentry if enabled, using settings
         if settings.SENTRY.ENABLED:
-            sentry_sdk.init(
-                dsn=settings.SENTRY.DSN,
-                enable_tracing=True,
-                release=settings.SENTRY.RELEASE,
-                environment=settings.SENTRY.ENVIRONMENT,
-                traces_sample_rate=settings.SENTRY.TRACES_SAMPLE_RATE,
-                profiles_sample_rate=settings.SENTRY.PROFILES_SAMPLE_RATE,
-                integrations=[AsyncioIntegration()],
-            )
+            initialize_sentry(integrations=[AsyncioIntegration()])
 
     def add_task(self, task: asyncio.Task[None]) -> None:
         """Track a new task"""

src/main.py

@@ -3,22 +3,16 @@ import re
 import uuid
 from collections.abc import Awaitable, Callable
 from contextlib import asynccontextmanager
-from typing import TYPE_CHECKING
 
 import sentry_sdk
 from fastapi import FastAPI, Request, Response
 from fastapi.middleware.cors import CORSMiddleware
 from fastapi.responses import JSONResponse
 from fastapi_pagination import add_pagination
-
-from src import prometheus
-from src.utils.logging import get_route_template
-
-if TYPE_CHECKING:
-    from sentry_sdk._types import Event, Hint
 from sentry_sdk.integrations.fastapi import FastApiIntegration
 from sentry_sdk.integrations.starlette import StarletteIntegration
 
+from src import prometheus
 from src.config import settings
 from src.db import engine, request_context
 from src.exceptions import HonchoException
@@ -31,6 +25,8 @@ from src.routers import (
     workspaces,
 )
 from src.security import create_admin_jwt
+from src.sentry import initialize_sentry
+from src.utils.logging import get_route_template
 
 
 def get_log_level() -> int:
@@ -71,27 +67,7 @@ async def setup_admin_jwt():
 # Sentry Setup
 SENTRY_ENABLED = settings.SENTRY.ENABLED
 if SENTRY_ENABLED:
-
-    def before_send(event: "Event", hint: "Hint") -> "Event | None":
-        if "exc_info" in hint:
-            _, exc_value, _ = hint["exc_info"]
-            # Filter out HonchoExceptions from being sent to Sentry
-            if isinstance(exc_value, HonchoException):
-                return None
-
-        return event
-
-    # Sentry SDK's default behavior:
-    # - Captures INFO+ level logs as breadcrumbs
-    # - Captures ERROR+ level logs as Sentry events
-    #
-    # For custom log levels, use the LoggingIntegration class:
-    # sentry_sdk.init(..., integrations=[LoggingIntegration(level=logging.INFO, event_level=logging.ERROR)])
-    sentry_sdk.init(
-        dsn=settings.SENTRY.DSN,
-        traces_sample_rate=settings.SENTRY.TRACES_SAMPLE_RATE,
-        profiles_sample_rate=settings.SENTRY.PROFILES_SAMPLE_RATE,
-        before_send=before_send,
+    initialize_sentry(
         integrations=[
             StarletteIntegration(
                 transaction_style="endpoint",
@@ -99,7 +75,7 @@ if SENTRY_ENABLED:
             FastApiIntegration(
                 transaction_style="endpoint",
             ),
-        ],
+        ]
     )
 
 

src/sentry.py

@@ -0,0 +1,58 @@
+"""Sentry initialization and configuration."""
+
+from __future__ import annotations
+
+from collections.abc import Sequence
+from typing import TYPE_CHECKING, Any
+
+import sentry_sdk
+from pydantic import ValidationError
+
+from src.config import settings
+from src.exceptions import HonchoException
+
+if TYPE_CHECKING:
+    from sentry_sdk._types import Event, Hint
+
+
+_UNSET = object()
+
+
+def _filter_sentry_event(event: Event, hint: Hint | None) -> Event | None:
+    """Filter out events raised from known non-actionable exceptions before Sentry sees them."""
+    if not hint:
+        return event
+
+    exc_info = hint.get("exc_info")
+    if not exc_info:
+        return event
+
+    _, exc_value, _ = exc_info
+    if isinstance(
+        exc_value, HonchoException | ValidationError
+    ):  # Filters out HonchoExceptions and ValidationErrors (typically coming from Pydantic)
+        return None
+
+    return event
+
+
+# Sentry SDK's default behavior:
+# - Captures INFO+ level logs as breadcrumbs
+# - Captures ERROR+ level logs as Sentry events
+#
+# For custom log levels, use the LoggingIntegration class:
+# sentry_sdk.init(..., integrations=[LoggingIntegration(level=logging.INFO, event_level=logging.ERROR)])
+def initialize_sentry(
+    *,
+    integrations: Sequence[Any],
+) -> None:
+    sentry_sdk.init(
+        dsn=settings.SENTRY.DSN,
+        enable_tracing=True,
+        release=settings.SENTRY.RELEASE,
+        environment=settings.SENTRY.ENVIRONMENT,
+        traces_sample_rate=settings.SENTRY.TRACES_SAMPLE_RATE,
+        profiles_sample_rate=settings.SENTRY.PROFILES_SAMPLE_RATE,
+        before_send=_filter_sentry_event,
+        integrations=integrations,
+    )