mirror of
https://github.com/home-assistant/core.git
synced 2026-04-02 00:20:30 +01:00
Mark send_message type hints as compulsory in notify (#160850)
This commit is contained in:
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from azure.servicebus import ServiceBusMessage
|
||||
from azure.servicebus.aio import ServiceBusClient, ServiceBusSender
|
||||
@@ -92,7 +93,7 @@ class ServiceBusNotificationService(BaseNotificationService):
|
||||
"""Initialize the service."""
|
||||
self._client = client
|
||||
|
||||
async def async_send_message(self, message, **kwargs):
|
||||
async def async_send_message(self, message: str, **kwargs: Any) -> None:
|
||||
"""Send a message."""
|
||||
dto = {ATTR_ASB_MESSAGE: message}
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
from webexpythonsdk import ApiError, WebexAPI, exceptions
|
||||
@@ -51,7 +52,7 @@ class CiscoWebexNotificationService(BaseNotificationService):
|
||||
self.room = room
|
||||
self.client = client
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a user."""
|
||||
|
||||
title = ""
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
from http import HTTPStatus
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
import voluptuous as vol
|
||||
@@ -81,7 +82,7 @@ class ClicksendNotificationService(BaseNotificationService):
|
||||
self.language = config[CONF_LANGUAGE]
|
||||
self.voice = config[CONF_VOICE]
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a voice call to a user."""
|
||||
data = {
|
||||
"messages": [
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components.notify import ATTR_TARGET, BaseNotificationService
|
||||
from homeassistant.core import HomeAssistant
|
||||
@@ -29,7 +30,7 @@ class DovadoSMSNotificationService(BaseNotificationService):
|
||||
"""Initialize the service."""
|
||||
self._client = client
|
||||
|
||||
def send_message(self, message, **kwargs):
|
||||
def send_message(self, message: str, **kwargs: Any) -> None:
|
||||
"""Send SMS to the specified target phone number."""
|
||||
if not (target := kwargs.get(ATTR_TARGET)):
|
||||
_LOGGER.error("One target is required")
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
from http import HTTPStatus
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
import voluptuous as vol
|
||||
@@ -46,7 +47,7 @@ class FacebookNotificationService(BaseNotificationService):
|
||||
"""Initialize the service."""
|
||||
self.page_access_token = access_token
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send some message."""
|
||||
payload = {"access_token": self.page_access_token}
|
||||
targets = kwargs.get(ATTR_TARGET)
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
import asyncio
|
||||
from http import HTTPStatus
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
@@ -47,7 +48,7 @@ class FlockNotificationService(BaseNotificationService):
|
||||
self._url = url
|
||||
self._session = session
|
||||
|
||||
async def async_send_message(self, message, **kwargs):
|
||||
async def async_send_message(self, message: str, **kwargs: Any) -> None:
|
||||
"""Send the message to the user."""
|
||||
payload = {"text": message}
|
||||
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from http import HTTPStatus
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from freesms import FreeClient
|
||||
import voluptuous as vol
|
||||
@@ -40,7 +41,7 @@ class FreeSMSNotificationService(BaseNotificationService):
|
||||
"""Initialize the service."""
|
||||
self.free_client = FreeClient(username, access_token)
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to the Free Mobile user cell."""
|
||||
resp = self.free_client.send_sms(message)
|
||||
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
from homeassistant.components.notify import (
|
||||
@@ -60,7 +62,7 @@ class HomematicNotificationService(BaseNotificationService):
|
||||
self.hass = hass
|
||||
self.data = data
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a notification to the device."""
|
||||
data = {**self.data, **kwargs.get(ATTR_DATA, {})}
|
||||
|
||||
|
||||
@@ -9,6 +9,7 @@ from http import HTTPStatus
|
||||
import json
|
||||
import logging
|
||||
import time
|
||||
from typing import Any
|
||||
from urllib.parse import urlparse
|
||||
import uuid
|
||||
|
||||
@@ -451,7 +452,7 @@ class HTML5NotificationService(BaseNotificationService):
|
||||
"""
|
||||
await self.hass.async_add_executor_job(partial(self.dismiss, **kwargs))
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a user."""
|
||||
tag = str(uuid.uuid4())
|
||||
payload = {
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pyjoin import get_devices, send_notification
|
||||
import voluptuous as vol
|
||||
@@ -66,7 +67,7 @@ class JoinNotificationService(BaseNotificationService):
|
||||
self._device_ids = device_ids
|
||||
self._device_names = device_names
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a user."""
|
||||
title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
|
||||
data = kwargs.get(ATTR_DATA) or {}
|
||||
|
||||
@@ -2,6 +2,8 @@
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
from typing import Any
|
||||
|
||||
from homeassistant.components.notify import ATTR_DATA, BaseNotificationService
|
||||
from homeassistant.core import HomeAssistant
|
||||
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
|
||||
@@ -27,7 +29,7 @@ class KebaNotificationService(BaseNotificationService):
|
||||
"""Initialize the service."""
|
||||
self._client = client
|
||||
|
||||
async def async_send_message(self, message="", **kwargs):
|
||||
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send the message."""
|
||||
text = message.replace(" ", "$") # Will be translated back by the display
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import aiohttp
|
||||
import jsonrpc_async
|
||||
@@ -93,7 +94,7 @@ class KodiNotificationService(BaseNotificationService):
|
||||
|
||||
self._server = jsonrpc_async.Server(self._url, **kwargs)
|
||||
|
||||
async def async_send_message(self, message="", **kwargs):
|
||||
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to Kodi."""
|
||||
try:
|
||||
data = kwargs.get(ATTR_DATA) or {}
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import socket
|
||||
from typing import Any
|
||||
from urllib.parse import urlencode
|
||||
|
||||
import voluptuous as vol
|
||||
@@ -73,7 +74,7 @@ class LannouncerNotificationService(BaseNotificationService):
|
||||
self._host = host
|
||||
self._port = port
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to Lannouncer."""
|
||||
data = kwargs.get(ATTR_DATA)
|
||||
if data is not None and ATTR_METHOD in data:
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from http import HTTPStatus
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
import voluptuous as vol
|
||||
@@ -56,7 +57,7 @@ class AutomateNotificationService(BaseNotificationService):
|
||||
self._recipient = recipient
|
||||
self._device = device
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a user."""
|
||||
|
||||
# Extract params from data dict
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from pymailgunner import (
|
||||
Client,
|
||||
@@ -91,7 +92,7 @@ class MailgunNotificationService(BaseNotificationService):
|
||||
return False
|
||||
return True
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a mail to the recipient."""
|
||||
|
||||
subject = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import messagebird
|
||||
from messagebird.client import ErrorException
|
||||
@@ -55,7 +56,7 @@ class MessageBirdNotificationService(BaseNotificationService):
|
||||
self.sender = sender
|
||||
self.client = client
|
||||
|
||||
def send_message(self, message=None, **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a specified target."""
|
||||
if not (targets := kwargs.get(ATTR_TARGET)):
|
||||
_LOGGER.error("No target specified")
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import pymsteams
|
||||
import voluptuous as vol
|
||||
@@ -49,7 +50,7 @@ class MSTeamsNotificationService(BaseNotificationService):
|
||||
"""Initialize the service."""
|
||||
self._webhook_url = webhook_url
|
||||
|
||||
def send_message(self, message=None, **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to the webhook."""
|
||||
|
||||
teams_message = pymsteams.connectorcard(self._webhook_url)
|
||||
|
||||
@@ -40,7 +40,7 @@ class NetgearNotifyService(BaseNotificationService):
|
||||
self.modem: Modem = discovery_info["modem"]
|
||||
discovery_info["entry"].async_on_unload(self.async_unregister_services)
|
||||
|
||||
async def async_send_message(self, message="", **kwargs):
|
||||
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a user."""
|
||||
|
||||
if not self.modem.token:
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
import logging
|
||||
import os.path
|
||||
from typing import Any
|
||||
|
||||
from notify_events import Message
|
||||
|
||||
@@ -123,7 +124,7 @@ class NotifyEventsNotificationService(BaseNotificationService):
|
||||
|
||||
return msg
|
||||
|
||||
def send_message(self, message, **kwargs):
|
||||
def send_message(self, message: str, **kwargs: Any) -> None:
|
||||
"""Send a message."""
|
||||
data = kwargs.get(ATTR_DATA) or {}
|
||||
token = data.get(ATTR_TOKEN, self.token)
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from http import HTTPStatus
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from rocketchat_API.APIExceptions.RocketExceptions import (
|
||||
RocketAuthenticationException,
|
||||
@@ -69,7 +70,7 @@ class RocketChatNotificationService(BaseNotificationService):
|
||||
self._room = room
|
||||
self._server = RocketChat(username, password, server_url=url)
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to Rocket.Chat."""
|
||||
data = kwargs.get(ATTR_DATA) or {}
|
||||
resp = self._server.chat_post_message(message, channel=self._room, **data)
|
||||
|
||||
@@ -4,6 +4,7 @@ from __future__ import annotations
|
||||
|
||||
from http import HTTPStatus
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from sendgrid import SendGridAPIClient
|
||||
import voluptuous as vol
|
||||
@@ -61,7 +62,7 @@ class SendgridNotificationService(BaseNotificationService):
|
||||
|
||||
self._sg = SendGridAPIClient(self.api_key)
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send an email to a user via SendGrid."""
|
||||
subject = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
from clx.xms.api import MtBatchTextSmsResult
|
||||
from clx.xms.client import Client
|
||||
@@ -67,7 +68,7 @@ class SinchNotificationService(BaseNotificationService):
|
||||
self.sender = config[CONF_SENDER]
|
||||
self.client = Client(config[CONF_SERVICE_PLAN_ID], config[CONF_API_KEY])
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a user."""
|
||||
targets = kwargs.get(ATTR_TARGET, self.default_recipients)
|
||||
data = kwargs.get(ATTR_DATA) or {}
|
||||
|
||||
@@ -5,6 +5,7 @@ from __future__ import annotations
|
||||
from http import HTTPStatus
|
||||
import json
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
import voluptuous as vol
|
||||
@@ -51,7 +52,7 @@ class SynologyChatNotificationService(BaseNotificationService):
|
||||
self._resource = resource
|
||||
self._verify_ssl = verify_ssl
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a user."""
|
||||
data = {"text": message}
|
||||
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import syslog
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
@@ -91,7 +92,7 @@ class SyslogNotificationService(BaseNotificationService):
|
||||
self._option = option
|
||||
self._priority = priority
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to syslog."""
|
||||
|
||||
title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
import urllib
|
||||
|
||||
from twilio.base.exceptions import TwilioRestException
|
||||
@@ -50,7 +51,7 @@ class TwilioCallNotificationService(BaseNotificationService):
|
||||
self.client = twilio_client
|
||||
self.from_number = from_number
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Call to specified target users."""
|
||||
if not (targets := kwargs.get(ATTR_TARGET)):
|
||||
_LOGGER.warning("At least 1 target is required")
|
||||
|
||||
@@ -3,6 +3,7 @@
|
||||
from __future__ import annotations
|
||||
|
||||
import logging
|
||||
from typing import Any
|
||||
|
||||
import voluptuous as vol
|
||||
|
||||
@@ -56,7 +57,7 @@ class TwilioSMSNotificationService(BaseNotificationService):
|
||||
self.client = twilio_client
|
||||
self.from_number = from_number
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send SMS to specified target user cell."""
|
||||
targets = kwargs.get(ATTR_TARGET)
|
||||
data = kwargs.get(ATTR_DATA) or {}
|
||||
|
||||
@@ -9,6 +9,7 @@ import json
|
||||
import logging
|
||||
import mimetypes
|
||||
import os
|
||||
from typing import Any
|
||||
|
||||
from TwitterAPI import TwitterAPI
|
||||
import voluptuous as vol
|
||||
@@ -79,7 +80,7 @@ class TwitterNotificationService(BaseNotificationService):
|
||||
consumer_key, consumer_secret, access_token_key, access_token_secret
|
||||
)
|
||||
|
||||
def send_message(self, message="", **kwargs):
|
||||
def send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Tweet a message, optionally with media."""
|
||||
data = kwargs.get(ATTR_DATA)
|
||||
targets = kwargs.get(ATTR_TARGET)
|
||||
|
||||
@@ -9,6 +9,7 @@ import mimetypes
|
||||
import pathlib
|
||||
import random
|
||||
import string
|
||||
from typing import Any
|
||||
|
||||
import requests
|
||||
import slixmpp
|
||||
@@ -101,7 +102,7 @@ class XmppNotificationService(BaseNotificationService):
|
||||
self._verify = verify
|
||||
self._room = room
|
||||
|
||||
async def async_send_message(self, message="", **kwargs):
|
||||
async def async_send_message(self, message: str = "", **kwargs: Any) -> None:
|
||||
"""Send a message to a user."""
|
||||
title = kwargs.get(ATTR_TITLE, ATTR_TITLE_DEFAULT)
|
||||
text = f"{title}: {message}" if title else message
|
||||
|
||||
@@ -2314,6 +2314,7 @@ _INHERITANCE_MATCH: dict[str, list[ClassTypeHintMatch]] = {
|
||||
kwargs_type="Any",
|
||||
return_type=None,
|
||||
has_async_counterpart=True,
|
||||
mandatory=True,
|
||||
),
|
||||
],
|
||||
),
|
||||
|
||||
Reference in New Issue
Block a user