39 lines
1.1 KiB
Python
39 lines
1.1 KiB
Python
from __future__ import annotations
|
|
|
|
import json
|
|
import socket
|
|
from dataclasses import asdict, dataclass
|
|
from datetime import datetime, timezone
|
|
from pathlib import Path
|
|
|
|
from .config import BackupConfig
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class ServerStatus:
|
|
reachable: bool
|
|
checked_at: str
|
|
message: str
|
|
|
|
|
|
def check_server(config: BackupConfig) -> ServerStatus:
|
|
checked_at = datetime.now(timezone.utc).isoformat()
|
|
if config.server_check.type != "tcp":
|
|
return ServerStatus(False, checked_at, f"Unsupported check type: {config.server_check.type}")
|
|
|
|
try:
|
|
with socket.create_connection(
|
|
(config.server_host, config.server_check.port),
|
|
timeout=config.server_check.timeout_seconds,
|
|
):
|
|
return ServerStatus(True, checked_at, "Server reachable")
|
|
except OSError as exc:
|
|
return ServerStatus(False, checked_at, str(exc))
|
|
|
|
|
|
def write_status(path: Path, status: ServerStatus, pending_count: int) -> None:
|
|
path.parent.mkdir(parents=True, exist_ok=True)
|
|
payload = asdict(status)
|
|
payload["pending_count"] = pending_count
|
|
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
|