94 lines
3.7 KiB
Python
94 lines
3.7 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Any, Dict, cast
|
|
|
|
import requests
|
|
|
|
from .errors import ApiError
|
|
from .models import *
|
|
|
|
|
|
class RoutesApiMixin:
|
|
# Routes
|
|
def routes_service(self, action: ServiceAction) -> CmdResult:
|
|
action_l = action.lower()
|
|
if action_l not in ("start", "stop", "restart"):
|
|
raise ValueError(f"Invalid action: {action}")
|
|
url = self._url("/api/v1/routes/service")
|
|
payload = {"action": action_l}
|
|
try:
|
|
# Короткий read-timeout: если systemctl завис на минуты, выходим,
|
|
# но backend продолжает выполнение (не привязан к request context).
|
|
resp = self._s.post(url, json=payload, timeout=(self.timeout, 2.0))
|
|
except requests.Timeout:
|
|
return CmdResult(
|
|
ok=True,
|
|
message=f"{action_l} accepted; backend is still running systemctl",
|
|
exit_code=None,
|
|
)
|
|
except requests.RequestException as e:
|
|
raise ApiError("API request failed", "POST", url, None, str(e)) from e
|
|
|
|
if not (200 <= resp.status_code < 300):
|
|
txt = resp.text.strip()
|
|
raise ApiError("API returned error", "POST", url, resp.status_code, txt)
|
|
|
|
data = cast(Dict[str, Any], self._json(resp) or {})
|
|
return self._parse_cmd_result(data)
|
|
|
|
def routes_clear(self) -> CmdResult:
|
|
data = cast(Dict[str, Any], self._json(self._request("POST", "/api/v1/routes/clear")) or {})
|
|
return self._parse_cmd_result(data)
|
|
|
|
def routes_cache_restore(self) -> CmdResult:
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(self._request("POST", "/api/v1/routes/cache/restore")) or {},
|
|
)
|
|
return self._parse_cmd_result(data)
|
|
|
|
def routes_precheck_debug(self, run_now: bool = True) -> CmdResult:
|
|
url = self._url("/api/v1/routes/precheck/debug")
|
|
payload = {"run_now": bool(run_now)}
|
|
try:
|
|
# Endpoint может запускать routes restart и выходить за 5s timeout.
|
|
# Для GUI считаем timeout признаком фонового принятого действия.
|
|
resp = self._s.post(url, json=payload, timeout=(self.timeout, 2.0))
|
|
except requests.Timeout:
|
|
return CmdResult(
|
|
ok=True,
|
|
message="precheck debug accepted; backend is still running",
|
|
exit_code=None,
|
|
)
|
|
except requests.RequestException as e:
|
|
raise ApiError("API request failed", "POST", url, None, str(e)) from e
|
|
|
|
if not (200 <= resp.status_code < 300):
|
|
txt = resp.text.strip()
|
|
raise ApiError("API returned error", "POST", url, resp.status_code, txt)
|
|
|
|
data = cast(Dict[str, Any], self._json(resp) or {})
|
|
return self._parse_cmd_result(data)
|
|
|
|
def routes_fix_policy_route(self) -> CmdResult:
|
|
data = cast(Dict[str, Any], self._json(self._request("POST", "/api/v1/routes/fix-policy-route")) or {})
|
|
return self._parse_cmd_result(data)
|
|
|
|
def routes_timer_get(self) -> RoutesTimerState:
|
|
data = cast(Dict[str, Any], self._json(self._request("GET", "/api/v1/routes/timer")) or {})
|
|
return RoutesTimerState(enabled=bool(data.get("enabled", False)))
|
|
|
|
def routes_timer_set(self, enabled: bool) -> CmdResult:
|
|
data = cast(
|
|
Dict[str, Any],
|
|
self._json(
|
|
self._request(
|
|
"POST",
|
|
"/api/v1/routes/timer",
|
|
json_body={"enabled": bool(enabled)},
|
|
)
|
|
)
|
|
or {},
|
|
)
|
|
return self._parse_cmd_result(data)
|