73 lines
2.4 KiB
Python
73 lines
2.4 KiB
Python
#!/usr/bin/env python3
|
|
from __future__ import annotations
|
|
|
|
from typing import List, cast
|
|
|
|
from api_client import (
|
|
DNSBenchmarkResponse,
|
|
DNSBenchmarkUpstream,
|
|
DNSStatus,
|
|
DNSUpstreamPoolState,
|
|
DnsUpstreams,
|
|
SmartdnsRuntimeState,
|
|
)
|
|
|
|
from .views import ActionView, ServiceAction
|
|
|
|
|
|
class DNSControllerMixin:
|
|
def dns_upstreams_view(self) -> DnsUpstreams:
|
|
return self.client.dns_upstreams_get()
|
|
|
|
def dns_upstreams_save(self, cfg: DnsUpstreams) -> None:
|
|
self.client.dns_upstreams_set(cfg)
|
|
|
|
def dns_upstream_pool_view(self) -> DNSUpstreamPoolState:
|
|
return self.client.dns_upstream_pool_get()
|
|
|
|
def dns_upstream_pool_save(self, items: List[DNSBenchmarkUpstream]) -> DNSUpstreamPoolState:
|
|
return self.client.dns_upstream_pool_set(items)
|
|
|
|
def dns_benchmark(
|
|
self,
|
|
upstreams: List[DNSBenchmarkUpstream],
|
|
domains: List[str],
|
|
timeout_ms: int = 1800,
|
|
attempts: int = 1,
|
|
concurrency: int = 6,
|
|
profile: str = "load",
|
|
) -> DNSBenchmarkResponse:
|
|
return self.client.dns_benchmark(
|
|
upstreams=upstreams,
|
|
domains=domains,
|
|
timeout_ms=timeout_ms,
|
|
attempts=attempts,
|
|
concurrency=concurrency,
|
|
profile=profile,
|
|
)
|
|
|
|
def dns_status_view(self) -> DNSStatus:
|
|
return self.client.dns_status_get()
|
|
|
|
def dns_mode_set(self, via: bool, smartdns_addr: str) -> DNSStatus:
|
|
return self.client.dns_mode_set(via, smartdns_addr)
|
|
|
|
def smartdns_service_action(self, action: str) -> DNSStatus:
|
|
act = action.strip().lower()
|
|
if act not in ("start", "stop", "restart"):
|
|
raise ValueError(f"Invalid SmartDNS action: {action}")
|
|
return self.client.dns_smartdns_service_set(cast(ServiceAction, act))
|
|
|
|
def smartdns_prewarm(self, limit: int = 0, aggressive_subs: bool = False) -> ActionView:
|
|
res = self.client.smartdns_prewarm(limit=limit, aggressive_subs=aggressive_subs)
|
|
return ActionView(ok=res.ok, pretty_text=self._pretty_cmd(res))
|
|
|
|
def smartdns_runtime_view(self) -> SmartdnsRuntimeState:
|
|
return self.client.smartdns_runtime_get()
|
|
|
|
def smartdns_runtime_set(self, enabled: bool, restart: bool = True) -> SmartdnsRuntimeState:
|
|
return self.client.smartdns_runtime_set(enabled=enabled, restart=restart)
|
|
|
|
# -------- Domains --------
|
|
|