feat(typing): Make monitor_config a TypedDict (#2931)

Author: sentrivana

sentry_sdk/_types.py

@@ -178,3 +178,37 @@
 
     BucketKey = Tuple[MetricType, str, MeasurementUnit, MetricTagsInternal]
     MetricMetaKey = Tuple[MetricType, str, MeasurementUnit]
+
+    MonitorConfigScheduleType = Literal["crontab", "interval"]
+    MonitorConfigScheduleUnit = Literal[
+        "year",
+        "month",
+        "week",
+        "day",
+        "hour",
+        "minute",
+        "second",  # not supported in Sentry and will result in a warning
+    ]
+
+    MonitorConfigSchedule = TypedDict(
+        "MonitorConfigSchedule",
+        {
+            "type": MonitorConfigScheduleType,
+            "value": Union[int, str],
+            "unit": MonitorConfigScheduleUnit,
+        },
+        total=False,
+    )
+
+    MonitorConfig = TypedDict(
+        "MonitorConfig",
+        {
+            "schedule": MonitorConfigSchedule,
+            "timezone": str,
+            "checkin_margin": int,
+            "max_runtime": int,
+            "failure_issue_threshold": int,
+            "recovery_threshold": int,
+        },
+        total=False,
+    )

sentry_sdk/crons/api.py

@@ -5,18 +5,18 @@
 
 
 if TYPE_CHECKING:
-    from typing import Any, Dict, Optional
-    from sentry_sdk._types import Event
+    from typing import Optional
+    from sentry_sdk._types import Event, MonitorConfig
 
 
 def _create_check_in_event(
-    monitor_slug=None,
-    check_in_id=None,
-    status=None,
-    duration_s=None,
-    monitor_config=None,
+    monitor_slug=None,  # type: Optional[str]
+    check_in_id=None,  # type: Optional[str]
+    status=None,  # type: Optional[str]
+    duration_s=None,  # type: Optional[float]
+    monitor_config=None,  # type: Optional[MonitorConfig]
 ):
-    # type: (Optional[str], Optional[str], Optional[str], Optional[float], Optional[Dict[str, Any]]) -> Event
+    # type: (...) -> Event
     options = Hub.current.client.options if Hub.current.client else {}
     check_in_id = check_in_id or uuid.uuid4().hex  # type: str
 
@@ -37,13 +37,13 @@ def _create_check_in_event(
 
 
 def capture_checkin(
-    monitor_slug=None,
-    check_in_id=None,
-    status=None,
-    duration=None,
-    monitor_config=None,
+    monitor_slug=None,  # type: Optional[str]
+    check_in_id=None,  # type: Optional[str]
+    status=None,  # type: Optional[str]
+    duration=None,  # type: Optional[float]
+    monitor_config=None,  # type: Optional[MonitorConfig]
 ):
-    # type: (Optional[str], Optional[str], Optional[str], Optional[float], Optional[Dict[str, Any]]) -> str
+    # type: (...) -> str
     check_in_event = _create_check_in_event(
         monitor_slug=monitor_slug,
         check_in_id=check_in_id,

sentry_sdk/crons/decorator.py

@@ -5,8 +5,9 @@
 from sentry_sdk.utils import now
 
 if TYPE_CHECKING:
-    from typing import Any, Optional, Type
+    from typing import Optional, Type
     from types import TracebackType
+    from sentry_sdk._types import MonitorConfig
 
 if PY2:
     from sentry_sdk.crons._decorator_py2 import MonitorMixin
@@ -48,7 +49,7 @@ def test(arg):
     """
 
     def __init__(self, monitor_slug=None, monitor_config=None):
-        # type: (Optional[str], Optional[dict[str, Any]]) -> None
+        # type: (Optional[str], Optional[MonitorConfig]) -> None
         self.monitor_slug = monitor_slug
         self.monitor_config = monitor_config
 

sentry_sdk/integrations/celery.py

@@ -3,6 +3,11 @@
 import sys
 import time
 
+try:
+    from typing import cast
+except ImportError:
+    cast = lambda _, o: o
+
 from sentry_sdk.api import continue_trace
 from sentry_sdk.consts import OP
 from sentry_sdk._compat import reraise
@@ -31,7 +36,15 @@
     from typing import Union
 
     from sentry_sdk.tracing import Span
-    from sentry_sdk._types import EventProcessor, Event, Hint, ExcInfo
+    from sentry_sdk._types import (
+        EventProcessor,
+        Event,
+        Hint,
+        ExcInfo,
+        MonitorConfig,
+        MonitorConfigScheduleType,
+        MonitorConfigScheduleUnit,
+    )
 
     F = TypeVar("F", bound=Callable[..., Any])
 
@@ -416,7 +429,7 @@ def _get_headers(task):
 
 
 def _get_humanized_interval(seconds):
-    # type: (float) -> Tuple[int, str]
+    # type: (float) -> Tuple[int, MonitorConfigScheduleUnit]
     TIME_UNITS = (  # noqa: N806
         ("day", 60 * 60 * 24.0),
         ("hour", 60 * 60.0),
@@ -427,17 +440,17 @@ def _get_humanized_interval(seconds):
     for unit, divider in TIME_UNITS:
         if seconds >= divider:
             interval = int(seconds / divider)
-            return (interval, unit)
+            return (interval, cast("MonitorConfigScheduleUnit", unit))
 
     return (int(seconds), "second")
 
 
 def _get_monitor_config(celery_schedule, app, monitor_name):
-    # type: (Any, Celery, str) -> Dict[str, Any]
-    monitor_config = {}  # type: Dict[str, Any]
-    schedule_type = None  # type: Optional[str]
+    # type: (Any, Celery, str) -> MonitorConfig
+    monitor_config = {}  # type: MonitorConfig
+    schedule_type = None  # type: Optional[MonitorConfigScheduleType]
     schedule_value = None  # type: Optional[Union[str, int]]
-    schedule_unit = None  # type: Optional[str]
+    schedule_unit = None  # type: Optional[MonitorConfigScheduleUnit]
 
     if isinstance(celery_schedule, crontab):
         schedule_type = "crontab"