mirror of
https://github.com/home-assistant/supervisor.git
synced 2025-10-17 15:50:04 +00:00
73849b7468
* Check management * Add test * Don't allow disable core_security * options and decorator * streamline config handling * streamline v2 * fix logging * Add tests * Fix test * cleanup v1 * fix api * Add more test * Expose option also for cli * address comments from Paulus * Address second comment * Update supervisor/resolution/checks/base.py Co-authored-by: Paulus Schoutsen <balloob@gmail.com> * fix lint * Fix black Co-authored-by: Pascal Vizeli <pvizeli@syshack.ch> Co-authored-by: Paulus Schoutsen <balloob@gmail.com>
101 lines
2.9 KiB
Python
101 lines
2.9 KiB
Python
"""Init file for Supervisor network RESTful API."""
|
|
import voluptuous as vol
|
|
|
|
from ..const import (
|
|
ATTR_ADDON,
|
|
ATTR_CONFIG,
|
|
ATTR_DISCOVERY,
|
|
ATTR_SERVICE,
|
|
ATTR_SERVICES,
|
|
ATTR_UUID,
|
|
REQUEST_FROM,
|
|
)
|
|
from ..coresys import CoreSysAttributes
|
|
from ..discovery.validate import valid_discovery_service
|
|
from ..exceptions import APIError, APIForbidden
|
|
from .utils import api_process, api_validate, require_home_assistant
|
|
|
|
SCHEMA_DISCOVERY = vol.Schema(
|
|
{
|
|
vol.Required(ATTR_SERVICE): valid_discovery_service,
|
|
vol.Optional(ATTR_CONFIG): vol.Maybe(dict),
|
|
}
|
|
)
|
|
|
|
|
|
class APIDiscovery(CoreSysAttributes):
|
|
"""Handle RESTful API for discovery functions."""
|
|
|
|
def _extract_message(self, request):
|
|
"""Extract discovery message from URL."""
|
|
message = self.sys_discovery.get(request.match_info.get("uuid"))
|
|
if not message:
|
|
raise APIError("Discovery message not found")
|
|
return message
|
|
|
|
@api_process
|
|
@require_home_assistant
|
|
async def list(self, request):
|
|
"""Show register services."""
|
|
|
|
# Get available discovery
|
|
discovery = []
|
|
for message in self.sys_discovery.list_messages:
|
|
discovery.append(
|
|
{
|
|
ATTR_ADDON: message.addon,
|
|
ATTR_SERVICE: message.service,
|
|
ATTR_UUID: message.uuid,
|
|
ATTR_CONFIG: message.config,
|
|
}
|
|
)
|
|
|
|
# Get available services/add-ons
|
|
services = {}
|
|
for addon in self.sys_addons.all:
|
|
for name in addon.discovery:
|
|
services.setdefault(name, []).append(addon.slug)
|
|
|
|
return {ATTR_DISCOVERY: discovery, ATTR_SERVICES: services}
|
|
|
|
@api_process
|
|
async def set_discovery(self, request):
|
|
"""Write data into a discovery pipeline."""
|
|
body = await api_validate(SCHEMA_DISCOVERY, request)
|
|
addon = request[REQUEST_FROM]
|
|
|
|
# Access?
|
|
if body[ATTR_SERVICE] not in addon.discovery:
|
|
raise APIForbidden("Can't use discovery!")
|
|
|
|
# Process discovery message
|
|
message = self.sys_discovery.send(addon, **body)
|
|
|
|
return {ATTR_UUID: message.uuid}
|
|
|
|
@api_process
|
|
@require_home_assistant
|
|
async def get_discovery(self, request):
|
|
"""Read data into a discovery message."""
|
|
message = self._extract_message(request)
|
|
|
|
return {
|
|
ATTR_ADDON: message.addon,
|
|
ATTR_SERVICE: message.service,
|
|
ATTR_UUID: message.uuid,
|
|
ATTR_CONFIG: message.config,
|
|
}
|
|
|
|
@api_process
|
|
async def del_discovery(self, request):
|
|
"""Delete data into a discovery message."""
|
|
message = self._extract_message(request)
|
|
addon = request[REQUEST_FROM]
|
|
|
|
# Permission
|
|
if message.addon != addon.slug:
|
|
raise APIForbidden("Can't remove discovery message")
|
|
|
|
self.sys_discovery.remove(message)
|
|
return True
|