api.py 1.6 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960
  1. import uuid
  2. import sentry_sdk
  3. from sentry_sdk.utils import logger
  4. from typing import TYPE_CHECKING
  5. if TYPE_CHECKING:
  6. from typing import Optional
  7. from sentry_sdk._types import Event, MonitorConfig
  8. def _create_check_in_event(
  9. monitor_slug: "Optional[str]" = None,
  10. check_in_id: "Optional[str]" = None,
  11. status: "Optional[str]" = None,
  12. duration_s: "Optional[float]" = None,
  13. monitor_config: "Optional[MonitorConfig]" = None,
  14. ) -> "Event":
  15. options = sentry_sdk.get_client().options
  16. check_in_id: str = check_in_id or uuid.uuid4().hex
  17. check_in: "Event" = {
  18. "type": "check_in",
  19. "monitor_slug": monitor_slug,
  20. "check_in_id": check_in_id,
  21. "status": status,
  22. "duration": duration_s,
  23. "environment": options.get("environment", None),
  24. "release": options.get("release", None),
  25. }
  26. if monitor_config:
  27. check_in["monitor_config"] = monitor_config
  28. return check_in
  29. def capture_checkin(
  30. monitor_slug: "Optional[str]" = None,
  31. check_in_id: "Optional[str]" = None,
  32. status: "Optional[str]" = None,
  33. duration: "Optional[float]" = None,
  34. monitor_config: "Optional[MonitorConfig]" = None,
  35. ) -> str:
  36. check_in_event = _create_check_in_event(
  37. monitor_slug=monitor_slug,
  38. check_in_id=check_in_id,
  39. status=status,
  40. duration_s=duration,
  41. monitor_config=monitor_config,
  42. )
  43. sentry_sdk.capture_event(check_in_event)
  44. logger.debug(
  45. f"[Crons] Captured check-in ({check_in_event.get('check_in_id')}): {check_in_event.get('monitor_slug')} -> {check_in_event.get('status')}"
  46. )
  47. return check_in_event["check_in_id"]