schema.py 4.4 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153
  1. from typing import Annotated
  2. from urllib.parse import urlparse
  3. from annotated_types import Ge, Le
  4. from django.conf import settings
  5. from django.core.exceptions import ValidationError as DjangoValidationError
  6. from django.core.validators import URLValidator
  7. from django.urls import reverse
  8. from ninja import ModelSchema
  9. from ninja.errors import ValidationError
  10. from pydantic import model_validator
  11. from glitchtip.schema import CamelSchema
  12. from .constants import HTTP_MONITOR_TYPES, MonitorType
  13. from .models import Monitor, MonitorCheck, StatusPage
  14. class MonitorCheckSchema(CamelSchema, ModelSchema):
  15. class Meta:
  16. model = MonitorCheck
  17. fields = ["is_up", "start_check", "reason"]
  18. class MonitorCheckResponseTimeSchema(MonitorCheckSchema, ModelSchema):
  19. """Monitor check with response time. Used in Monitors detail api and monitor checks list"""
  20. class Meta(MonitorCheckSchema.Meta):
  21. fields = MonitorCheckSchema.Meta.fields + ["response_time"]
  22. class MonitorIn(CamelSchema, ModelSchema):
  23. expected_body: str
  24. expected_status: int | None
  25. timeout: Annotated[int, Ge(1), Le(60)] | None
  26. project: str | None = None
  27. @model_validator(mode="after")
  28. def validate(self):
  29. monitor_type = self.monitor_type
  30. if self.url == "" and monitor_type in HTTP_MONITOR_TYPES + (MonitorType.SSL,):
  31. raise ValidationError("URL is required for " + monitor_type)
  32. if monitor_type in HTTP_MONITOR_TYPES:
  33. try:
  34. URLValidator()(self.url)
  35. except DjangoValidationError as err:
  36. raise ValidationError("Invalid Url") from err
  37. if self.expected_status is None and monitor_type in [
  38. MonitorType.GET,
  39. MonitorType.POST,
  40. ]:
  41. raise ValidationError("Expected status is required for " + monitor_type)
  42. if monitor_type == MonitorType.PORT:
  43. url = self.url.replace("http://", "//", 1)
  44. if not url.startswith("//"):
  45. url = "//" + url
  46. parsed_url = urlparse(url)
  47. message = "Invalid Port URL, expected hostname and port"
  48. try:
  49. if not all([parsed_url.hostname, parsed_url.port]):
  50. raise ValidationError(message)
  51. except ValueError as err:
  52. raise ValidationError(message) from err
  53. self.url = f"{parsed_url.hostname}:{parsed_url.port}"
  54. return self
  55. class Meta:
  56. model = Monitor
  57. fields = [
  58. "monitor_type",
  59. "name",
  60. "url",
  61. "interval",
  62. ]
  63. class MonitorSchema(CamelSchema, ModelSchema):
  64. project_id: str | None
  65. environment_id: int | None
  66. is_up: bool | None
  67. last_change: str | None
  68. heartbeat_endpoint: str | None
  69. project_name: str | None = None
  70. env_name: str | None = None
  71. checks: list[MonitorCheckSchema]
  72. organization_id: int
  73. monitor_type: MonitorType
  74. class Meta:
  75. model = Monitor
  76. fields = [
  77. "id",
  78. "monitor_type",
  79. "endpoint_id",
  80. "created",
  81. "name",
  82. "url",
  83. "expected_status",
  84. "expected_body",
  85. "interval",
  86. "timeout",
  87. ]
  88. class Config(CamelSchema.Config):
  89. coerce_numbers_to_str = True
  90. @staticmethod
  91. def resolve_is_up(obj):
  92. return obj.latest_is_up
  93. @staticmethod
  94. def resolve_last_change(obj):
  95. if obj.last_change:
  96. return obj.last_change.isoformat().replace("+00:00", "Z")
  97. @staticmethod
  98. def resolve_heartbeat_endpoint(obj):
  99. if obj.endpoint_id:
  100. return settings.GLITCHTIP_URL.geturl() + reverse(
  101. "api:heartbeat_check",
  102. kwargs={
  103. "organization_slug": obj.organization.slug,
  104. "endpoint_id": obj.endpoint_id,
  105. },
  106. )
  107. @staticmethod
  108. def resolve_project_name(obj):
  109. if obj.project:
  110. return obj.project.name
  111. class MonitorDetailSchema(MonitorSchema):
  112. checks: list[MonitorCheckResponseTimeSchema]
  113. class StatusPageIn(CamelSchema, ModelSchema):
  114. is_public: bool = False
  115. class Meta:
  116. model = StatusPage
  117. fields = ["name", "is_public"]
  118. class StatusPageSchema(StatusPageIn, ModelSchema):
  119. monitors: list[MonitorSchema]
  120. class Meta(StatusPageIn.Meta):
  121. fields = ["name", "slug", "is_public"]