mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-11-21 00:36:57 +00:00
f6faa18409
It seems that the codebase is not formatted with the latest ruff version. This PR reformats the codebase with ruff 0.5.7.
118 lines
3.7 KiB
Python
118 lines
3.7 KiB
Python
"""Init file for Supervisor DNS RESTful API."""
|
|
|
|
import asyncio
|
|
from collections.abc import Awaitable
|
|
import logging
|
|
from typing import Any
|
|
|
|
from aiohttp import web
|
|
import voluptuous as vol
|
|
|
|
from ..const import (
|
|
ATTR_BLK_READ,
|
|
ATTR_BLK_WRITE,
|
|
ATTR_CPU_PERCENT,
|
|
ATTR_HOST,
|
|
ATTR_LOCALS,
|
|
ATTR_MEMORY_LIMIT,
|
|
ATTR_MEMORY_PERCENT,
|
|
ATTR_MEMORY_USAGE,
|
|
ATTR_NETWORK_RX,
|
|
ATTR_NETWORK_TX,
|
|
ATTR_SERVERS,
|
|
ATTR_UPDATE_AVAILABLE,
|
|
ATTR_VERSION,
|
|
ATTR_VERSION_LATEST,
|
|
)
|
|
from ..coresys import CoreSysAttributes
|
|
from ..exceptions import APIError
|
|
from ..validate import dns_server_list, version_tag
|
|
from .const import ATTR_FALLBACK, ATTR_LLMNR, ATTR_MDNS
|
|
from .utils import api_process, api_validate
|
|
|
|
_LOGGER: logging.Logger = logging.getLogger(__name__)
|
|
|
|
# pylint: disable=no-value-for-parameter
|
|
SCHEMA_OPTIONS = vol.Schema(
|
|
{
|
|
vol.Optional(ATTR_SERVERS): dns_server_list,
|
|
vol.Optional(ATTR_FALLBACK): vol.Boolean(),
|
|
}
|
|
)
|
|
|
|
SCHEMA_VERSION = vol.Schema({vol.Optional(ATTR_VERSION): version_tag})
|
|
|
|
|
|
class APICoreDNS(CoreSysAttributes):
|
|
"""Handle RESTful API for DNS functions."""
|
|
|
|
@api_process
|
|
async def info(self, request: web.Request) -> dict[str, Any]:
|
|
"""Return DNS information."""
|
|
return {
|
|
ATTR_VERSION: self.sys_plugins.dns.version,
|
|
ATTR_VERSION_LATEST: self.sys_plugins.dns.latest_version,
|
|
ATTR_UPDATE_AVAILABLE: self.sys_plugins.dns.need_update,
|
|
ATTR_HOST: str(self.sys_docker.network.dns),
|
|
ATTR_SERVERS: self.sys_plugins.dns.servers,
|
|
ATTR_LOCALS: self.sys_plugins.dns.locals,
|
|
ATTR_MDNS: self.sys_plugins.dns.mdns,
|
|
ATTR_LLMNR: self.sys_plugins.dns.llmnr,
|
|
ATTR_FALLBACK: self.sys_plugins.dns.fallback,
|
|
}
|
|
|
|
@api_process
|
|
async def options(self, request: web.Request) -> None:
|
|
"""Set DNS options."""
|
|
body = await api_validate(SCHEMA_OPTIONS, request)
|
|
restart_required = False
|
|
|
|
if ATTR_SERVERS in body:
|
|
self.sys_plugins.dns.servers = body[ATTR_SERVERS]
|
|
restart_required = True
|
|
|
|
if ATTR_FALLBACK in body:
|
|
self.sys_plugins.dns.fallback = body[ATTR_FALLBACK]
|
|
restart_required = True
|
|
|
|
if restart_required:
|
|
self.sys_create_task(self.sys_plugins.dns.restart())
|
|
|
|
self.sys_plugins.dns.save_data()
|
|
|
|
@api_process
|
|
async def stats(self, request: web.Request) -> dict[str, Any]:
|
|
"""Return resource information."""
|
|
stats = await self.sys_plugins.dns.stats()
|
|
|
|
return {
|
|
ATTR_CPU_PERCENT: stats.cpu_percent,
|
|
ATTR_MEMORY_USAGE: stats.memory_usage,
|
|
ATTR_MEMORY_LIMIT: stats.memory_limit,
|
|
ATTR_MEMORY_PERCENT: stats.memory_percent,
|
|
ATTR_NETWORK_RX: stats.network_rx,
|
|
ATTR_NETWORK_TX: stats.network_tx,
|
|
ATTR_BLK_READ: stats.blk_read,
|
|
ATTR_BLK_WRITE: stats.blk_write,
|
|
}
|
|
|
|
@api_process
|
|
async def update(self, request: web.Request) -> None:
|
|
"""Update DNS plugin."""
|
|
body = await api_validate(SCHEMA_VERSION, request)
|
|
version = body.get(ATTR_VERSION, self.sys_plugins.dns.latest_version)
|
|
|
|
if version == self.sys_plugins.dns.version:
|
|
raise APIError(f"Version {version} is already in use")
|
|
await asyncio.shield(self.sys_plugins.dns.update(version))
|
|
|
|
@api_process
|
|
def restart(self, request: web.Request) -> Awaitable[None]:
|
|
"""Restart CoreDNS plugin."""
|
|
return asyncio.shield(self.sys_plugins.dns.restart())
|
|
|
|
@api_process
|
|
def reset(self, request: web.Request) -> Awaitable[None]:
|
|
"""Reset CoreDNS plugin."""
|
|
return asyncio.shield(self.sys_plugins.dns.reset())
|