Add initial implementation

This commit is contained in:
sqozz 2022-10-29 06:18:28 +02:00
commit 2e10115449
4 changed files with 134 additions and 0 deletions

1
__init__.py Normal file
View File

@ -0,0 +1 @@
"""The ntfy component."""

5
const.py Normal file
View File

@ -0,0 +1,5 @@
"""Constants for ntfy component."""
DEFAULT_VERIFY_SSL = True
DEFAULT_NAME = "ntfy"
ATTR_DEFAULT_TOPIC = "homeassistant"

8
manifest.json Normal file
View File

@ -0,0 +1,8 @@
{
"domain": "ntfy",
"name": "ntfy",
"documentation": "https://www.home-assistant.io/integrations/rest",
"requirements": [],
"codeowners": ["sqozz"],
"iot_class": "cloud_push"
}

120
notify.py Normal file
View File

@ -0,0 +1,120 @@
"""ntfy platform for notify component."""
import logging
from http import HTTPStatus
import homeassistant.helpers.config_validation as cv
import requests
import voluptuous as vol
from homeassistant.components.notify import (ATTR_TARGET, ATTR_TITLE,
PLATFORM_SCHEMA,
BaseNotificationService)
from homeassistant.const import (CONF_NAME, CONF_PASSWORD, CONF_URL,
CONF_USERNAME, CONF_VERIFY_SSL,
HTTP_BASIC_AUTHENTICATION)
from .const import DEFAULT_NAME, DEFAULT_VERIFY_SSL, ATTR_DEFAULT_TOPIC
from homeassistant.helpers.template import Template
PLATFORM_SCHEMA = PLATFORM_SCHEMA.extend(
{
vol.Required(CONF_URL): cv.url,
vol.Optional(CONF_NAME, default=DEFAULT_NAME): cv.string,
vol.Optional(CONF_PASSWORD): cv.string,
vol.Optional(CONF_USERNAME): cv.string,
vol.Optional(CONF_VERIFY_SSL, default=DEFAULT_VERIFY_SSL): cv.boolean,
}
)
_LOGGER = logging.getLogger(__name__)
async def async_get_service(hass, config, discovery_info=None):
"""Get the ntfy notification service."""
ntfy_url = config.get(CONF_URL)
username = config.get(CONF_USERNAME)
password = config.get(CONF_PASSWORD)
verify_ssl = config.get(CONF_VERIFY_SSL)
if username and password:
auth = requests.auth.HTTPBasicAuth(username, password)
else:
auth = None
return NtfyNotificationService(
hass,
ntfy_url,
auth,
verify_ssl,
)
class NtfyNotificationService(BaseNotificationService):
"""Implementation of a notification service for ntfy."""
def __init__(
self,
hass,
ntfy_url,
auth,
verify_ssl,
):
"""Initialize the service."""
self._ntfy_url = ntfy_url
self._hass = hass
self._auth = auth
self._verify_ssl = verify_ssl
def send_message(self, message="", **kwargs):
"""Send a message to a topic."""
data = {"message": message}
# Target is a list as of 0.29 and we don't want to break existing
# integrations, so just return the first target in the list.
data["topic"] = kwargs.get(ATTR_TARGET, [ATTR_DEFAULT_TOPIC])[0]
if ATTR_TITLE in kwargs:
data["title"] = kwargs[ATTR_TITLE]
if "data" in kwargs:
if "tags" in kwargs["data"]:
data["tags"] = list(kwargs["data"]["tags"])
if "priority" in kwargs["data"]:
prio = int(kwargs["data"]["priority"])
if not prio in range(1, 6):
raise ValueError("priority must be between 1-5")
data["priority"] = prio
if "actions" in kwargs["data"]:
data["actions"] = list(kwargs["data"]["actions"])
for attr in ["click", "attach", "filename", "delay", "email"]:
if attr in kwargs["data"]:
data[attr] = str(kwargs["data"][attr])
response = requests.post(
self._ntfy_url,
json=data,
timeout=10,
auth=self._auth,
verify=self._verify_ssl,
)
if (
response.status_code >= HTTPStatus.INTERNAL_SERVER_ERROR
and response.status_code < 600
):
_LOGGER.exception(
"Server error. Response %d: %s:", response.status_code, response.reason
)
elif (
response.status_code >= HTTPStatus.BAD_REQUEST
and response.status_code < HTTPStatus.INTERNAL_SERVER_ERROR
):
_LOGGER.exception(
"Client error. Response %d: %s:", response.status_code, response.reason
)
elif (
response.status_code >= HTTPStatus.OK
and response.status_code < HTTPStatus.MULTIPLE_CHOICES
):
_LOGGER.debug(
"Success. Response %d: %s:", response.status_code, response.reason
)
else:
_LOGGER.debug("Response %d: %s:", response.status_code, response.reason)