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

Add hints to get_service in azure service bus (#86694)

Browse Source
This commit is contained in:
epenet 2023-01-26 17:20:52 +01:00 committed by GitHub
parent eb5d63237c
commit b3380261d7
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
1 changed files with 16 additions and 7 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

23
homeassistant/components/azure_service_bus/notify.py
View File

@ -1,9 +1,11 @@
"""Support for azure service bus notification."""
from __future__ import annotations
import json
import logging
from azure.servicebus import ServiceBusMessage
from azure.servicebus.aio import ServiceBusClient
from azure.servicebus.aio import ServiceBusClient, ServiceBusSender
from azure.servicebus.exceptions import (
MessagingEntityNotFoundError,
ServiceBusConnectionError,
@ -19,7 +21,9 @@ from homeassistant.components.notify import (
BaseNotificationService,
)
from homeassistant.const import CONTENT_TYPE_JSON
from homeassistant.core import HomeAssistant
import homeassistant.helpers.config_validation as cv
from homeassistant.helpers.typing import ConfigType, DiscoveryInfoType
CONF_CONNECTION_STRING = "connection_string"
CONF_QUEUE_NAME = "queue"
@ -47,11 +51,15 @@ PLATFORM_SCHEMA = vol.All(
_LOGGER = logging.getLogger(__name__)
def get_service(hass, config, discovery_info=None):
def get_service(
hass: HomeAssistant,
config: ConfigType,
discovery_info: DiscoveryInfoType | None = None,
) -> ServiceBusNotificationService | None:
"""Get the notification service."""
connection_string = config[CONF_CONNECTION_STRING]
queue_name = config.get(CONF_QUEUE_NAME)
topic_name = config.get(CONF_TOPIC_NAME)
connection_string: str = config[CONF_CONNECTION_STRING]
queue_name: str | None = config.get(CONF_QUEUE_NAME)
topic_name: str | None = config.get(CONF_TOPIC_NAME)
# Library can do synchronous IO when creating the clients.
# Passes in loop here, but can't run setup on the event loop.
@ -59,10 +67,11 @@ def get_service(hass, config, discovery_info=None):
connection_string, loop=hass.loop
)
client: ServiceBusSender | None = None
try:
if queue_name:
client = servicebus.get_queue_sender(queue_name)
else:
elif topic_name:
client = servicebus.get_topic_sender(topic_name)
except (ServiceBusConnectionError, MessagingEntityNotFoundError) as err:
_LOGGER.error(
@ -72,7 +81,7 @@ def get_service(hass, config, discovery_info=None):
)
return None
return ServiceBusNotificationService(client)
return ServiceBusNotificationService(client) if client else None
class ServiceBusNotificationService(BaseNotificationService):