55 lines
1.8 KiB
Python
55 lines
1.8 KiB
Python
from dataclasses import dataclass
|
|
from enum import StrEnum
|
|
|
|
|
|
class MatchType(StrEnum):
|
|
STATUS = "status" # следим только за HTTP-статусом
|
|
BODY = "body" # тело ответа должно совпадать целиком
|
|
JSON = "json" # конкретное поле JSON должно быть равно эталону
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Service:
|
|
id: int
|
|
name: str
|
|
url: str
|
|
match_type: MatchType
|
|
ok_status: int
|
|
ok_body: str
|
|
json_path: str | None
|
|
json_value: str | None
|
|
is_up: bool
|
|
created_at: str
|
|
# Сколько проверок подряд провалилось. Инцидент открывается, только когда
|
|
# счётчик дорастает до monitor.failure_threshold, — см. env.py.
|
|
fail_streak: int = 0
|
|
# Пауза: сервис остаётся в списке, но не пингуется и не будит ночью.
|
|
# Нужно на время плановых работ, чтобы не удалять сервис со статистикой.
|
|
is_paused: bool = False
|
|
last_latency_ms: int | None = None
|
|
last_checked_at: str | None = None
|
|
|
|
@property
|
|
def rule(self) -> str:
|
|
if self.match_type is MatchType.STATUS:
|
|
return f"HTTP-статус = {self.ok_status}"
|
|
if self.match_type is MatchType.JSON:
|
|
return f"<code>{self.json_path}</code> = <code>{self.json_value}</code>"
|
|
return "тело ответа неизменно"
|
|
|
|
@property
|
|
def mark(self) -> str:
|
|
if self.is_paused:
|
|
return "⏸"
|
|
return "🟢" if self.is_up else "🔴"
|
|
|
|
|
|
@dataclass(slots=True)
|
|
class Incident:
|
|
id: int
|
|
service_id: int
|
|
started_at: str
|
|
ended_at: str | None
|
|
reason: str
|
|
last_reminder_at: str
|