data.py

Go to the documentation of this file.

"""Support for RESTful API."""
 
from __future__ import annotations
 
import logging
import ssl
 
import httpx
import xmltodict
 
from homeassistant.core import HomeAssistant
from homeassistant.helpers import template
from homeassistant.helpers.httpx_client import create_async_httpx_client
from homeassistant.helpers.json import json_dumps
from homeassistant.util.ssl import SSLCipherList
 
from .const import XML_MIME_TYPES
 
DEFAULT_TIMEOUT = 10
 
_LOGGER = logging.getLogger(__name__)
 
 
class RestData:
    """Class for handling the data retrieval."""
 
    def __init__(
        self,
        hass: HomeAssistant,
        method: str,
        resource: str,
        encoding: str,
        auth: httpx.DigestAuth | tuple[str, str] | None,
        headers: dict[str, str] | None,
        params: dict[str, str] | None,
        data: str | None,
        verify_ssl: bool,
        ssl_cipher_list: str,
        timeout: int = DEFAULT_TIMEOUT,
    ) -> None:
        """Initialize the data object."""
        self._hass_hass = hass
        self._method_method = method
        self._resource_resource = resource
        self._encoding_encoding = encoding
        self._auth_auth = auth
        self._headers_headers = headers
        self._params_params = params
        self._request_data_request_data = data
        self._timeout_timeout = timeout
        self._verify_ssl_verify_ssl = verify_ssl
        self._ssl_cipher_list_ssl_cipher_list = SSLCipherList(ssl_cipher_list)
        self._async_client_async_client: httpx.AsyncClient | None = None
        self.datadata: str | None = None
        self.last_exceptionlast_exception: Exception | None = None
        self.headersheaders: httpx.Headers | None = None
 
    def set_payload(self, payload: str) -> None:
        """Set request data."""
        self._request_data_request_data = payload
 
    @property
    def url(self) -> str:
        """Get url."""
        return self._resource_resource
 
    def set_url(self, url: str) -> None:
        """Set url."""
        self._resource_resource = url
 
    def data_without_xml(self) -> str | None:
        """If the data is an XML string, convert it to a JSON string."""
        _LOGGER.debug("Data fetched from resource: %s", self.datadata)
        if (
            (value := self.datadata) is not None
            # If the http request failed, headers will be None
            and (headers := self.headersheaders) is not None
            and (content_type := headers.get("content-type"))
            and content_type.startswith(XML_MIME_TYPES)
        ):
            value = json_dumps(xmltodict.parse(value))
            _LOGGER.debug("JSON converted from XML: %s", value)
        return value
 
    async def async_update(self, log_errors: bool = True) -> None:
        """Get the latest data from REST service with provided method."""
        if not self._async_client_async_client:
            self._async_client_async_client = create_async_httpx_client(
                self._hass_hass,
                verify_ssl=self._verify_ssl_verify_ssl,
                default_encoding=self._encoding_encoding,
                ssl_cipher_list=self._ssl_cipher_list_ssl_cipher_list,
            )
 
        rendered_headers = template.render_complex(self._headers_headers, parse_result=False)
        rendered_params = template.render_complex(self._params_params)
 
        _LOGGER.debug("Updating from %s", self._resource_resource)
        try:
            response = await self._async_client_async_client.request(
                self._method_method,
                self._resource_resource,
                headers=rendered_headers,
                params=rendered_params,
                auth=self._auth_auth,
                content=self._request_data_request_data,
                timeout=self._timeout_timeout,
                follow_redirects=True,
            )
            self.datadata = response.text
            self.headersheaders = response.headers
        except httpx.TimeoutException as ex:
            if log_errors:
                _LOGGER.error("Timeout while fetching data: %s", self._resource_resource)
            self.last_exceptionlast_exception = ex
            self.datadata = None
            self.headersheaders = None
        except httpx.RequestError as ex:
            if log_errors:
                _LOGGER.error(
                    "Error fetching data: %s failed with %s", self._resource_resource, ex
                )
            self.last_exceptionlast_exception = ex
            self.datadata = None
            self.headersheaders = None
        except ssl.SSLError as ex:
            if log_errors:
                _LOGGER.error(
                    "Error connecting to %s failed with %s", self._resource_resource, ex
                )
            self.last_exceptionlast_exception = ex
            self.datadata = None
            self.headersheaders = None