Break out UniFi get_unifi_api function (#111323)

Break out get api function
This commit is contained in:
Robert Svensson 2024-02-27 18:20:06 +01:00 committed by GitHub
parent 6bd7a74fab
commit 4281f648d2
No known key found for this signature in database
GPG Key ID: B5690EEEBB952194
3 changed files with 96 additions and 89 deletions

View File

@ -1,3 +1,4 @@
"""Internal functionality not part of HA infrastructure.""" """Internal functionality not part of HA infrastructure."""
from .hub import UnifiHub, get_unifi_api # noqa: F401 from .api import get_unifi_api # noqa: F401
from .hub import UnifiHub # noqa: F401

View File

@ -0,0 +1,92 @@
"""Provide an object to communicate with UniFi Network application."""
from __future__ import annotations
import asyncio
import ssl
from types import MappingProxyType
from typing import Any, Literal
from aiohttp import CookieJar
import aiounifi
from aiounifi.models.configuration import Configuration
from homeassistant.const import (
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONF_VERIFY_SSL,
)
from homeassistant.core import HomeAssistant
from homeassistant.helpers import aiohttp_client
from ..const import CONF_SITE_ID, LOGGER
from ..errors import AuthenticationRequired, CannotConnect
async def get_unifi_api(
hass: HomeAssistant,
config: MappingProxyType[str, Any],
) -> aiounifi.Controller:
"""Create a aiounifi object and verify authentication."""
ssl_context: ssl.SSLContext | Literal[False] = False
if verify_ssl := config.get(CONF_VERIFY_SSL):
session = aiohttp_client.async_get_clientsession(hass)
if isinstance(verify_ssl, str):
ssl_context = ssl.create_default_context(cafile=verify_ssl)
else:
session = aiohttp_client.async_create_clientsession(
hass, verify_ssl=False, cookie_jar=CookieJar(unsafe=True)
)
api = aiounifi.Controller(
Configuration(
session,
host=config[CONF_HOST],
username=config[CONF_USERNAME],
password=config[CONF_PASSWORD],
port=config[CONF_PORT],
site=config[CONF_SITE_ID],
ssl_context=ssl_context,
)
)
try:
async with asyncio.timeout(10):
await api.login()
return api
except aiounifi.Unauthorized as err:
LOGGER.warning(
"Connected to UniFi Network at %s but not registered: %s",
config[CONF_HOST],
err,
)
raise AuthenticationRequired from err
except (
TimeoutError,
aiounifi.BadGateway,
aiounifi.Forbidden,
aiounifi.ServiceUnavailable,
aiounifi.RequestError,
aiounifi.ResponseError,
) as err:
LOGGER.error(
"Error connecting to the UniFi Network at %s: %s", config[CONF_HOST], err
)
raise CannotConnect from err
except aiounifi.LoginRequired as err:
LOGGER.warning(
"Connected to UniFi Network at %s but login required: %s",
config[CONF_HOST],
err,
)
raise AuthenticationRequired from err
except aiounifi.AiounifiException as err:
LOGGER.exception("Unknown UniFi Network communication error occurred: %s", err)
raise AuthenticationRequired from err

View File

@ -1,35 +1,18 @@
"""UniFi Network abstraction.""" """UniFi Network abstraction."""
from __future__ import annotations from __future__ import annotations
import asyncio
from collections.abc import Iterable from collections.abc import Iterable
from datetime import datetime, timedelta from datetime import datetime, timedelta
from functools import partial from functools import partial
import ssl
from types import MappingProxyType
from typing import Any, Literal
from aiohttp import CookieJar
import aiounifi import aiounifi
from aiounifi.interfaces.api_handlers import ItemEvent from aiounifi.interfaces.api_handlers import ItemEvent
from aiounifi.models.configuration import Configuration
from aiounifi.models.device import DeviceSetPoePortModeRequest from aiounifi.models.device import DeviceSetPoePortModeRequest
from homeassistant.config_entries import ConfigEntry from homeassistant.config_entries import ConfigEntry
from homeassistant.const import ( from homeassistant.const import CONF_HOST, Platform
CONF_HOST,
CONF_PASSWORD,
CONF_PORT,
CONF_USERNAME,
CONF_VERIFY_SSL,
Platform,
)
from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback from homeassistant.core import CALLBACK_TYPE, Event, HomeAssistant, callback
from homeassistant.helpers import ( from homeassistant.helpers import device_registry as dr, entity_registry as er
aiohttp_client,
device_registry as dr,
entity_registry as er,
)
from homeassistant.helpers.device_registry import ( from homeassistant.helpers.device_registry import (
DeviceEntry, DeviceEntry,
DeviceEntryType, DeviceEntryType,
@ -67,12 +50,10 @@ from ..const import (
DEFAULT_TRACK_DEVICES, DEFAULT_TRACK_DEVICES,
DEFAULT_TRACK_WIRED_CLIENTS, DEFAULT_TRACK_WIRED_CLIENTS,
DOMAIN as UNIFI_DOMAIN, DOMAIN as UNIFI_DOMAIN,
LOGGER,
PLATFORMS, PLATFORMS,
UNIFI_WIRELESS_CLIENTS, UNIFI_WIRELESS_CLIENTS,
) )
from ..entity import UnifiEntity, UnifiEntityDescription from ..entity import UnifiEntity, UnifiEntityDescription
from ..errors import AuthenticationRequired, CannotConnect
from .websocket import UnifiWebsocket from .websocket import UnifiWebsocket
CHECK_HEARTBEAT_INTERVAL = timedelta(seconds=1) CHECK_HEARTBEAT_INTERVAL = timedelta(seconds=1)
@ -418,70 +399,3 @@ class UnifiHub:
self._cancel_poe_command = None self._cancel_poe_command = None
return True return True
async def get_unifi_api(
hass: HomeAssistant,
config: MappingProxyType[str, Any],
) -> aiounifi.Controller:
"""Create a aiounifi object and verify authentication."""
ssl_context: ssl.SSLContext | Literal[False] = False
if verify_ssl := config.get(CONF_VERIFY_SSL):
session = aiohttp_client.async_get_clientsession(hass)
if isinstance(verify_ssl, str):
ssl_context = ssl.create_default_context(cafile=verify_ssl)
else:
session = aiohttp_client.async_create_clientsession(
hass, verify_ssl=False, cookie_jar=CookieJar(unsafe=True)
)
api = aiounifi.Controller(
Configuration(
session,
host=config[CONF_HOST],
username=config[CONF_USERNAME],
password=config[CONF_PASSWORD],
port=config[CONF_PORT],
site=config[CONF_SITE_ID],
ssl_context=ssl_context,
)
)
try:
async with asyncio.timeout(10):
await api.login()
return api
except aiounifi.Unauthorized as err:
LOGGER.warning(
"Connected to UniFi Network at %s but not registered: %s",
config[CONF_HOST],
err,
)
raise AuthenticationRequired from err
except (
TimeoutError,
aiounifi.BadGateway,
aiounifi.Forbidden,
aiounifi.ServiceUnavailable,
aiounifi.RequestError,
aiounifi.ResponseError,
) as err:
LOGGER.error(
"Error connecting to the UniFi Network at %s: %s", config[CONF_HOST], err
)
raise CannotConnect from err
except aiounifi.LoginRequired as err:
LOGGER.warning(
"Connected to UniFi Network at %s but login required: %s",
config[CONF_HOST],
err,
)
raise AuthenticationRequired from err
except aiounifi.AiounifiException as err:
LOGGER.exception("Unknown UniFi Network communication error occurred: %s", err)
raise AuthenticationRequired from err