Use HTTPStatus instead of HTTP_* int constants in mobile_app responses (#56418)
This commit is contained in:
parent
26faac0567
commit
8e18ca3b6e
2 changed files with 18 additions and 14 deletions
|
@ -2,6 +2,7 @@
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
|
|
||||||
from collections.abc import Callable
|
from collections.abc import Callable
|
||||||
|
from http import HTTPStatus
|
||||||
import json
|
import json
|
||||||
import logging
|
import logging
|
||||||
|
|
||||||
|
@ -9,12 +10,7 @@ from aiohttp.web import Response, json_response
|
||||||
from nacl.encoding import Base64Encoder
|
from nacl.encoding import Base64Encoder
|
||||||
from nacl.secret import SecretBox
|
from nacl.secret import SecretBox
|
||||||
|
|
||||||
from homeassistant.const import (
|
from homeassistant.const import ATTR_DEVICE_ID, CONTENT_TYPE_JSON
|
||||||
ATTR_DEVICE_ID,
|
|
||||||
CONTENT_TYPE_JSON,
|
|
||||||
HTTP_BAD_REQUEST,
|
|
||||||
HTTP_OK,
|
|
||||||
)
|
|
||||||
from homeassistant.core import Context, HomeAssistant
|
from homeassistant.core import Context, HomeAssistant
|
||||||
from homeassistant.helpers.entity import DeviceInfo
|
from homeassistant.helpers.entity import DeviceInfo
|
||||||
from homeassistant.helpers.json import JSONEncoder
|
from homeassistant.helpers.json import JSONEncoder
|
||||||
|
@ -95,7 +91,9 @@ def registration_context(registration: dict) -> Context:
|
||||||
return Context(user_id=registration[CONF_USER_ID])
|
return Context(user_id=registration[CONF_USER_ID])
|
||||||
|
|
||||||
|
|
||||||
def empty_okay_response(headers: dict = None, status: int = HTTP_OK) -> Response:
|
def empty_okay_response(
|
||||||
|
headers: dict = None, status: HTTPStatus = HTTPStatus.OK
|
||||||
|
) -> Response:
|
||||||
"""Return a Response with empty JSON object and a 200."""
|
"""Return a Response with empty JSON object and a 200."""
|
||||||
return Response(
|
return Response(
|
||||||
text="{}", status=status, content_type=CONTENT_TYPE_JSON, headers=headers
|
text="{}", status=status, content_type=CONTENT_TYPE_JSON, headers=headers
|
||||||
|
@ -103,7 +101,10 @@ def empty_okay_response(headers: dict = None, status: int = HTTP_OK) -> Response
|
||||||
|
|
||||||
|
|
||||||
def error_response(
|
def error_response(
|
||||||
code: str, message: str, status: int = HTTP_BAD_REQUEST, headers: dict = None
|
code: str,
|
||||||
|
message: str,
|
||||||
|
status: HTTPStatus = HTTPStatus.BAD_REQUEST,
|
||||||
|
headers: dict = None,
|
||||||
) -> Response:
|
) -> Response:
|
||||||
"""Return an error Response."""
|
"""Return an error Response."""
|
||||||
return json_response(
|
return json_response(
|
||||||
|
@ -147,7 +148,11 @@ def savable_state(hass: HomeAssistant) -> dict:
|
||||||
|
|
||||||
|
|
||||||
def webhook_response(
|
def webhook_response(
|
||||||
data, *, registration: dict, status: int = HTTP_OK, headers: dict = None
|
data,
|
||||||
|
*,
|
||||||
|
registration: dict,
|
||||||
|
status: HTTPStatus = HTTPStatus.OK,
|
||||||
|
headers: dict = None,
|
||||||
) -> Response:
|
) -> Response:
|
||||||
"""Return a encrypted response if registration supports it."""
|
"""Return a encrypted response if registration supports it."""
|
||||||
data = json.dumps(data, cls=JSONEncoder)
|
data = json.dumps(data, cls=JSONEncoder)
|
||||||
|
|
|
@ -2,6 +2,7 @@
|
||||||
import asyncio
|
import asyncio
|
||||||
from contextlib import suppress
|
from contextlib import suppress
|
||||||
from functools import wraps
|
from functools import wraps
|
||||||
|
from http import HTTPStatus
|
||||||
import logging
|
import logging
|
||||||
import secrets
|
import secrets
|
||||||
|
|
||||||
|
@ -30,8 +31,6 @@ from homeassistant.const import (
|
||||||
ATTR_SERVICE_DATA,
|
ATTR_SERVICE_DATA,
|
||||||
ATTR_SUPPORTED_FEATURES,
|
ATTR_SUPPORTED_FEATURES,
|
||||||
CONF_WEBHOOK_ID,
|
CONF_WEBHOOK_ID,
|
||||||
HTTP_BAD_REQUEST,
|
|
||||||
HTTP_CREATED,
|
|
||||||
)
|
)
|
||||||
from homeassistant.core import EventOrigin, HomeAssistant
|
from homeassistant.core import EventOrigin, HomeAssistant
|
||||||
from homeassistant.exceptions import HomeAssistantError, ServiceNotFound
|
from homeassistant.exceptions import HomeAssistantError, ServiceNotFound
|
||||||
|
@ -158,7 +157,7 @@ async def handle_webhook(
|
||||||
req_data = await request.json()
|
req_data = await request.json()
|
||||||
except ValueError:
|
except ValueError:
|
||||||
_LOGGER.warning("Received invalid JSON from mobile_app device: %s", device_name)
|
_LOGGER.warning("Received invalid JSON from mobile_app device: %s", device_name)
|
||||||
return empty_okay_response(status=HTTP_BAD_REQUEST)
|
return empty_okay_response(status=HTTPStatus.BAD_REQUEST)
|
||||||
|
|
||||||
if (
|
if (
|
||||||
ATTR_WEBHOOK_ENCRYPTED not in req_data
|
ATTR_WEBHOOK_ENCRYPTED not in req_data
|
||||||
|
@ -265,7 +264,7 @@ async def webhook_stream_camera(hass, config_entry, data):
|
||||||
return webhook_response(
|
return webhook_response(
|
||||||
{"success": False},
|
{"success": False},
|
||||||
registration=config_entry.data,
|
registration=config_entry.data,
|
||||||
status=HTTP_BAD_REQUEST,
|
status=HTTPStatus.BAD_REQUEST,
|
||||||
)
|
)
|
||||||
|
|
||||||
resp = {"mjpeg_path": f"/api/camera_proxy_stream/{camera.entity_id}"}
|
resp = {"mjpeg_path": f"/api/camera_proxy_stream/{camera.entity_id}"}
|
||||||
|
@ -435,7 +434,7 @@ async def webhook_register_sensor(hass, config_entry, data):
|
||||||
return webhook_response(
|
return webhook_response(
|
||||||
{"success": True},
|
{"success": True},
|
||||||
registration=config_entry.data,
|
registration=config_entry.data,
|
||||||
status=HTTP_CREATED,
|
status=HTTPStatus.CREATED,
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
|
|
Loading…
Add table
Add a link
Reference in a new issue