jeans/core
jeans/core
1
0
Fork 0
mirror of https://github.com/home-assistant/core.git synced 2025-07-22 20:57:21 +00:00
Code Issues Packages Projects Releases Wiki Activity

Add websocket type hints in blueprint (#80531)

Browse Source
This commit is contained in:
epenet 2022-10-19 04:23:17 +02:00 committed by GitHub
parent 971ac015e7
commit c0be1d9fad
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
2 changed files with 29 additions and 18 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
homeassistant/components/blueprint/models.py
View File

@ -51,7 +51,7 @@ class Blueprint:
def __init__( def __init__(
self, self,
data: dict, data: dict[str, Any],
*, *,
path: str | None = None, path: str | None = None,
expected_domain: str | None = None, expected_domain: str | None = None,

45
homeassistant/components/blueprint/websocket_api.py
View File

@ -1,6 +1,8 @@
"""Websocket API for blueprint.""" """Websocket API for blueprint."""
from __future__ import annotations from __future__ import annotations
from typing import Any, cast
import async_timeout import async_timeout
import voluptuous as vol import voluptuous as vol
@ -31,12 +33,14 @@ def async_setup(hass: HomeAssistant):
} }
) )
@websocket_api.async_response @websocket_api.async_response
async def ws_list_blueprints(hass, connection, msg): async def ws_list_blueprints(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""List available blueprints.""" """List available blueprints."""
domain_blueprints: dict[str, models.DomainBlueprints] | None = hass.data.get( domain_blueprints: dict[str, models.DomainBlueprints] = hass.data.get(DOMAIN, {})
DOMAIN, {} results: dict[str, Any] = {}
)
results = {}
if msg["domain"] not in domain_blueprints: if msg["domain"] not in domain_blueprints:
connection.send_result(msg["id"], results) connection.send_result(msg["id"], results)
@ -62,7 +66,11 @@ async def ws_list_blueprints(hass, connection, msg):
} }
) )
@websocket_api.async_response @websocket_api.async_response
async def ws_import_blueprint(hass, connection, msg): async def ws_import_blueprint(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Import a blueprint.""" """Import a blueprint."""
async with async_timeout.timeout(10): async with async_timeout.timeout(10):
imported_blueprint = await importer.fetch_blueprint_from_url(hass, msg["url"]) imported_blueprint = await importer.fetch_blueprint_from_url(hass, msg["url"])
@ -96,15 +104,17 @@ async def ws_import_blueprint(hass, connection, msg):
} }
) )
@websocket_api.async_response @websocket_api.async_response
async def ws_save_blueprint(hass, connection, msg): async def ws_save_blueprint(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Save a blueprint.""" """Save a blueprint."""
path = msg["path"] path = msg["path"]
domain = msg["domain"] domain = msg["domain"]
domain_blueprints: dict[str, models.DomainBlueprints] | None = hass.data.get( domain_blueprints: dict[str, models.DomainBlueprints] = hass.data.get(DOMAIN, {})
DOMAIN, {}
)
if domain not in domain_blueprints: if domain not in domain_blueprints:
connection.send_error( connection.send_error(
@ -112,9 +122,8 @@ async def ws_save_blueprint(hass, connection, msg):
) )
try: try:
blueprint = models.Blueprint( yaml_data = cast(dict[str, Any], yaml.parse_yaml(msg["yaml"]))
yaml.parse_yaml(msg["yaml"]), expected_domain=domain blueprint = models.Blueprint(yaml_data, expected_domain=domain)
)
if "source_url" in msg: if "source_url" in msg:
blueprint.update_metadata(source_url=msg["source_url"]) blueprint.update_metadata(source_url=msg["source_url"])
except HomeAssistantError as err: except HomeAssistantError as err:
@ -143,15 +152,17 @@ async def ws_save_blueprint(hass, connection, msg):
} }
) )
@websocket_api.async_response @websocket_api.async_response
async def ws_delete_blueprint(hass, connection, msg): async def ws_delete_blueprint(
hass: HomeAssistant,
connection: websocket_api.ActiveConnection,
msg: dict[str, Any],
) -> None:
"""Delete a blueprint.""" """Delete a blueprint."""
path = msg["path"] path = msg["path"]
domain = msg["domain"] domain = msg["domain"]
domain_blueprints: dict[str, models.DomainBlueprints] | None = hass.data.get( domain_blueprints: dict[str, models.DomainBlueprints] = hass.data.get(DOMAIN, {})
DOMAIN, {}
)
if domain not in domain_blueprints: if domain not in domain_blueprints:
connection.send_error( connection.send_error(