69 lines
2.1 KiB
Python
69 lines
2.1 KiB
Python
from datetime import datetime, timezone
|
|
from enum import StrEnum
|
|
from typing import Self
|
|
from uuid import UUID
|
|
|
|
from pydantic import BaseModel, ConfigDict, Field, HttpUrl, model_validator
|
|
|
|
|
|
class State(StrEnum):
|
|
UNKNOWN = "unknown"
|
|
UP = "up"
|
|
DOWN = "down"
|
|
ERROR = "error"
|
|
|
|
|
|
class CurrentStatus(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
state: State = State.UNKNOWN
|
|
checked_at: datetime | None = None
|
|
latency_ms: float | None = Field(default=None, ge=0)
|
|
http_status: int | None = Field(default=None, ge=100, le=599)
|
|
final_url: str | None = None
|
|
error: str | None = None
|
|
|
|
|
|
class MonitorCreate(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
name: str = Field(min_length=1, max_length=120)
|
|
url: HttpUrl
|
|
|
|
@model_validator(mode="after")
|
|
def safe_url_shape(self) -> Self:
|
|
if self.url.scheme not in {"http", "https"}:
|
|
raise ValueError("only http and https URLs are supported")
|
|
if self.url.username is not None or self.url.password is not None:
|
|
raise ValueError("URL credentials are not allowed")
|
|
return self
|
|
|
|
|
|
class MonitorUpdate(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
name: str | None = Field(default=None, min_length=1, max_length=120)
|
|
url: HttpUrl | None = None
|
|
|
|
@model_validator(mode="after")
|
|
def validate_update(self) -> Self:
|
|
if self.name is None and self.url is None:
|
|
raise ValueError("at least one field is required")
|
|
if self.url is not None:
|
|
if self.url.scheme not in {"http", "https"}:
|
|
raise ValueError("only http and https URLs are supported")
|
|
if self.url.username is not None or self.url.password is not None:
|
|
raise ValueError("URL credentials are not allowed")
|
|
return self
|
|
|
|
|
|
class Monitor(BaseModel):
|
|
model_config = ConfigDict(extra="forbid")
|
|
id: UUID
|
|
name: str
|
|
url: HttpUrl
|
|
created_at: datetime
|
|
updated_at: datetime
|
|
current_status: CurrentStatus
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(timezone.utc)
|