diff --git a/sentry_sdk/_types.py b/sentry_sdk/_types.py index 49bffb3416..91208e51d4 100644 --- a/sentry_sdk/_types.py +++ b/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, + ) diff --git a/sentry_sdk/crons/api.py b/sentry_sdk/crons/api.py index 92d113a924..1a95583301 100644 --- a/sentry_sdk/crons/api.py +++ b/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, diff --git a/sentry_sdk/crons/decorator.py b/sentry_sdk/crons/decorator.py index 5bedcb48b0..6c5f747b97 100644 --- a/sentry_sdk/crons/decorator.py +++ b/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 diff --git a/sentry_sdk/integrations/celery.py b/sentry_sdk/integrations/celery.py index f2e1aff48a..984197316f 100644 --- a/sentry_sdk/integrations/celery.py +++ b/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"