mirror of
https://github.com/home-assistant/core.git
synced 2025-07-22 20:57:21 +00:00
Improve type hints in webhook implementations (#121588)
This commit is contained in:
parent
7aada39b77
commit
9ca398ef82
@ -89,7 +89,9 @@ async def async_setup(hass: HomeAssistant, hass_config: ConfigType) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
async def handle_webhook(hass, webhook_id, request):
|
||||
async def handle_webhook(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> web.Response:
|
||||
"""Handle incoming webhook from Geofency."""
|
||||
try:
|
||||
data = WEBHOOK_SCHEMA(dict(await request.post()))
|
||||
|
@ -55,7 +55,9 @@ WEBHOOK_SCHEMA = vol.Schema(
|
||||
)
|
||||
|
||||
|
||||
async def handle_webhook(hass, webhook_id, request):
|
||||
async def handle_webhook(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> web.Response:
|
||||
"""Handle incoming webhook with GPSLogger request."""
|
||||
try:
|
||||
data = WEBHOOK_SCHEMA(dict(await request.post()))
|
||||
|
@ -61,7 +61,9 @@ WEBHOOK_SCHEMA = vol.All(
|
||||
)
|
||||
|
||||
|
||||
async def handle_webhook(hass, webhook_id, request):
|
||||
async def handle_webhook(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> web.Response:
|
||||
"""Handle incoming webhook from Locative."""
|
||||
try:
|
||||
data = WEBHOOK_SCHEMA(dict(await request.post()))
|
||||
|
@ -5,6 +5,7 @@ import hmac
|
||||
import json
|
||||
import logging
|
||||
|
||||
from aiohttp import web
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components import webhook
|
||||
@ -48,13 +49,15 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
async def handle_webhook(hass, webhook_id, request):
|
||||
async def handle_webhook(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> None:
|
||||
"""Handle incoming webhook with Mailgun inbound messages."""
|
||||
body = await request.text()
|
||||
try:
|
||||
data = json.loads(body) if body else {}
|
||||
except ValueError:
|
||||
return None
|
||||
return
|
||||
|
||||
if (
|
||||
isinstance(data, dict)
|
||||
|
@ -38,7 +38,7 @@ async def async_handle_webhook(
|
||||
data = await request.json()
|
||||
except ValueError as err:
|
||||
_LOGGER.error("Error in data: %s", err)
|
||||
return None
|
||||
return
|
||||
|
||||
_LOGGER.debug("Got webhook data: %s", data)
|
||||
|
||||
|
@ -5,7 +5,7 @@ import json
|
||||
import logging
|
||||
import re
|
||||
|
||||
from aiohttp.web import json_response
|
||||
from aiohttp import web
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components import cloud, mqtt, webhook
|
||||
@ -153,7 +153,9 @@ async def async_connect_mqtt(hass, component):
|
||||
return True
|
||||
|
||||
|
||||
async def handle_webhook(hass, webhook_id, request):
|
||||
async def handle_webhook(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> web.Response:
|
||||
"""Handle webhook callback.
|
||||
|
||||
iOS sets the "topic" as part of the payload.
|
||||
@ -166,7 +168,7 @@ async def handle_webhook(hass, webhook_id, request):
|
||||
message = await request.json()
|
||||
except ValueError:
|
||||
_LOGGER.warning("Received invalid JSON from OwnTracks")
|
||||
return json_response([])
|
||||
return web.json_response([])
|
||||
|
||||
# Android doesn't populate topic
|
||||
if "topic" not in message:
|
||||
@ -183,7 +185,7 @@ async def handle_webhook(hass, webhook_id, request):
|
||||
" set a username in Connection -> Identification"
|
||||
)
|
||||
# Keep it as a 200 response so the incorrect packet is discarded
|
||||
return json_response([])
|
||||
return web.json_response([])
|
||||
|
||||
async_dispatcher_send(hass, DOMAIN, hass, context, message)
|
||||
|
||||
@ -200,7 +202,7 @@ async def handle_webhook(hass, webhook_id, request):
|
||||
]
|
||||
|
||||
if message["_type"] == "encrypted" and context.secret:
|
||||
return json_response(
|
||||
return web.json_response(
|
||||
{
|
||||
"_type": "encrypted",
|
||||
"data": encrypt_message(
|
||||
@ -209,7 +211,7 @@ async def handle_webhook(hass, webhook_id, request):
|
||||
}
|
||||
)
|
||||
|
||||
return json_response(response)
|
||||
return web.json_response(response)
|
||||
|
||||
|
||||
class OwnTracksContext:
|
||||
|
@ -184,7 +184,9 @@ async def _async_update_listener(hass: HomeAssistant, entry: ConfigEntry) -> Non
|
||||
await hass.config_entries.async_reload(entry.entry_id)
|
||||
|
||||
|
||||
async def handle_webhook(hass, webhook_id, request):
|
||||
async def handle_webhook(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> web.Response | None:
|
||||
"""Handle incoming webhook from Plaato."""
|
||||
try:
|
||||
data = WEBHOOK_SCHEMA(await request.json())
|
||||
|
@ -3,6 +3,7 @@
|
||||
import asyncio
|
||||
import logging
|
||||
|
||||
from aiohttp import web
|
||||
from httpx import ConnectTimeout
|
||||
from pypoint import PointSession
|
||||
import voluptuous as vol
|
||||
@ -158,13 +159,15 @@ async def async_unload_entry(hass: HomeAssistant, entry: ConfigEntry) -> bool:
|
||||
return unload_ok
|
||||
|
||||
|
||||
async def handle_webhook(hass, webhook_id, request):
|
||||
async def handle_webhook(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> None:
|
||||
"""Handle webhook callback."""
|
||||
try:
|
||||
data = await request.json()
|
||||
_LOGGER.debug("Webhook %s: %s", webhook_id, data)
|
||||
except ValueError:
|
||||
return None
|
||||
return
|
||||
|
||||
if isinstance(data, dict):
|
||||
data["webhook_id"] = webhook_id
|
||||
|
@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
import logging
|
||||
import secrets
|
||||
|
||||
from aiohttp import web
|
||||
from toonapi import Status, Toon, ToonError
|
||||
|
||||
from homeassistant.components import cloud, webhook
|
||||
@ -98,7 +99,7 @@ class ToonDataUpdateCoordinator(DataUpdateCoordinator[Status]):
|
||||
)
|
||||
|
||||
async def handle_webhook(
|
||||
self, hass: HomeAssistant, webhook_id: str, request
|
||||
self, hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> None:
|
||||
"""Handle webhook callback."""
|
||||
try:
|
||||
|
@ -56,7 +56,9 @@ WEBHOOK_SCHEMA = vol.Schema(
|
||||
)
|
||||
|
||||
|
||||
async def handle_webhook(hass, webhook_id, request):
|
||||
async def handle_webhook(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> web.Response:
|
||||
"""Handle incoming webhook with Traccar Client request."""
|
||||
try:
|
||||
data = WEBHOOK_SCHEMA(dict(request.query))
|
||||
|
@ -46,7 +46,9 @@ async def async_setup(hass: HomeAssistant, config: ConfigType) -> bool:
|
||||
return True
|
||||
|
||||
|
||||
async def handle_webhook(hass, webhook_id, request):
|
||||
async def handle_webhook(
|
||||
hass: HomeAssistant, webhook_id: str, request: web.Request
|
||||
) -> web.Response:
|
||||
"""Handle incoming webhook from Twilio for inbound messages and calls."""
|
||||
data = dict(await request.post())
|
||||
data["webhook_id"] = webhook_id
|
||||
|
Loading…
x
Reference in New Issue
Block a user