81 lines
2.2 KiB
Python
81 lines
2.2 KiB
Python
from datetime import UTC, datetime, timedelta
|
|
from typing import Any
|
|
|
|
from motor.motor_asyncio import AsyncIOMotorCollection
|
|
|
|
from app.core.config import settings
|
|
from app.core.mongodb import get_database
|
|
from app.models.health import HealthSnapshot
|
|
|
|
COLLECTION_NAME = "health_history"
|
|
|
|
|
|
def utcnow() -> datetime:
|
|
return datetime.now(UTC)
|
|
|
|
|
|
async def collection() -> AsyncIOMotorCollection[dict[str, Any]]:
|
|
database = await get_database()
|
|
return database[COLLECTION_NAME]
|
|
|
|
|
|
async def ensure_health_indexes() -> None:
|
|
samples = await collection()
|
|
await samples.create_index(
|
|
[("proxy_port", 1), ("timestamp", -1)],
|
|
name="ix_health_proxy_port_timestamp",
|
|
)
|
|
await samples.create_index(
|
|
"expires_at",
|
|
expireAfterSeconds=0,
|
|
name="ttl_health_history",
|
|
)
|
|
|
|
|
|
def deserialize_health_sample(raw: dict[str, Any] | None) -> HealthSnapshot | None:
|
|
if raw is None:
|
|
return None
|
|
document = dict(raw)
|
|
document.pop("_id", None)
|
|
document.pop("expires_at", None)
|
|
return HealthSnapshot.model_validate(document)
|
|
|
|
|
|
async def record_health_sample(snapshot: HealthSnapshot) -> HealthSnapshot:
|
|
samples = await collection()
|
|
document = snapshot.model_dump(mode="python")
|
|
document["expires_at"] = document["timestamp"] + timedelta(
|
|
seconds=settings.HEALTH_HISTORY_TTL_SECONDS,
|
|
)
|
|
await samples.insert_one(document)
|
|
return snapshot
|
|
|
|
|
|
async def get_latest_health_sample(proxy_port: int) -> HealthSnapshot | None:
|
|
samples = await collection()
|
|
raw = await samples.find_one(
|
|
{"proxy_port": proxy_port},
|
|
sort=[("timestamp", -1)],
|
|
)
|
|
return deserialize_health_sample(raw)
|
|
|
|
|
|
async def list_health_history(
|
|
proxy_port: int,
|
|
limit: int = 40,
|
|
) -> list[HealthSnapshot]:
|
|
samples = await collection()
|
|
cursor = samples.find({"proxy_port": proxy_port}).sort("timestamp", -1).limit(limit)
|
|
history: list[HealthSnapshot] = []
|
|
async for raw in cursor:
|
|
snapshot = deserialize_health_sample(raw)
|
|
if snapshot is not None:
|
|
history.append(snapshot)
|
|
history.reverse()
|
|
return history
|
|
|
|
|
|
async def delete_health_history(proxy_port: int) -> None:
|
|
samples = await collection()
|
|
await samples.delete_many({"proxy_port": proxy_port})
|