Add exception to catch failed calls

custom_components/polestar_api/__init__.py

@@ -5,6 +5,8 @@
 
 from aiohttp import ClientConnectionError
 from async_timeout import timeout
+
+from .pypolestar.exception import PolestarApiException
 from .pypolestar.polestar import PolestarApi
 from .polestar import Polestar
 from homeassistant.config_entries import ConfigEntry
@@ -42,14 +44,17 @@ async def async_setup_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> b
     _LOGGER.debug("async_setup_entry: %s", config_entry)
     polestarApi = Polestar(
         hass, conf[CONF_USERNAME], conf[CONF_PASSWORD])
-    await polestarApi.init()
+    try:
+        await polestarApi.init()
 
-    hass.data.setdefault(DOMAIN, {})
-    hass.data[DOMAIN][config_entry.entry_id] = polestarApi
+        hass.data.setdefault(DOMAIN, {})
+        hass.data[DOMAIN][config_entry.entry_id] = polestarApi
 
-    await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
+        await hass.config_entries.async_forward_entry_setups(config_entry, PLATFORMS)
 
-    return True
+        return True
+    except PolestarApiException as e:
+        _LOGGER.exception("API Exception %s", str(e))
 
 
 async def async_unload_entry(hass: HomeAssistant, config_entry: ConfigEntry) -> bool:

custom_components/polestar_api/polestar.py

@@ -1,8 +1,11 @@
 
+from datetime import timedelta
 import logging
 
 from .pypolestar.polestar import PolestarApi
+
 from urllib3 import disable_warnings
+
 from homeassistant.core import HomeAssistant
 
 POST_HEADER_JSON = {"Content-Type": "application/json"}
@@ -35,6 +38,10 @@ def get_latest_data(self, query: str, field_name: str):
         return self.polestarApi.get_latest_data(query, field_name)
 
     def get_latest_call_code(self):
+        # if AUTH code last code is not 200 then we return that error code,
+        # otherwise just give the call_code in API
+        if self.polestarApi.auth.latest_call_code != 200:
+            return self.polestarApi.auth.latest_call_code
         return self.polestarApi.latest_call_code
 
     async def async_update(self) -> None:

custom_components/polestar_api/pypolestar/auth.py

@@ -4,6 +4,8 @@
 
 from datetime import datetime, timedelta
 
+from .exception import PolestarAuthException
+
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -38,8 +40,8 @@ async def get_token(self) -> None:
         result = await self._client_session.get("https://pc-api.polestar.com/eu-north-1/auth/", params=params, headers=headers)
         self.latest_call_code = result.status_code
         if result.status_code != 200:
-            _LOGGER.error(f"Error getting token {result.status_code}")
-            return
+            raise PolestarAuthException(
+                f"Error getting token", result.status_code)
         resultData = result.json()
         _LOGGER.debug(resultData)
 
@@ -80,8 +82,8 @@ async def _get_code(self) -> None:
         )
         self.latest_call_code = result.status_code
         if result.status_code != 302:
-            _LOGGER.error(f"Error getting code {result.status_code}")
-            return
+            raise PolestarAuthException(
+                f"Error getting code", result.status_code)
 
         # get the realUrl
         url = result.url
@@ -92,8 +94,9 @@ async def _get_code(self) -> None:
         self.latest_call_code = result.status_code
 
         if result.status_code != 200:
-            _LOGGER.error(f"Error getting code callback {result.status_code}")
-            return
+            raise PolestarAuthException(
+                f"Error getting code callback", result.status_code)
+
         # url encode the code
         result = await self._client_session.get(url)
         self.latest_call_code = result.status_code
@@ -109,6 +112,6 @@ async def _get_resume_path(self):
         }
         result = await self._client_session.get("https://polestarid.eu.polestar.com/as/authorization.oauth2", params=params)
         if result.status_code != 303:
-            _LOGGER.error(f"Error getting resume path {result.status_code}")
-            return
+            raise PolestarAuthException(
+                f"Error getting resume path ", result.status_code)
         return result.next_request.url.params

custom_components/polestar_api/pypolestar/exception.py

@@ -0,0 +1,21 @@
+class PolestarApiException(Exception):
+    """Base class for exceptions in this module."""
+
+
+class PolestarAuthException(Exception):
+    """Base class for exceptions in Auth module."""
+    error_code: int = None
+    message: str = None
+
+    def __init__(self, message, error_code) -> None:
+        super().__init__(message)
+        self.error_code = error_code
+        return None
+
+
+class PolestarNotAuthorizedException(Exception):
+    """Exception for unauthorized call"""
+
+
+class PolestarNoDataException(Exception):
+    """Exception for no data"""

custom_components/polestar_api/pypolestar/polestar.py

@@ -2,6 +2,8 @@
 import httpx
 
 from datetime import datetime, timedelta
+
+from .exception import PolestarApiException, PolestarAuthException, PolestarNoDataException, PolestarNotAuthorizedException
 from .auth import PolestarAuth
 from .const import CACHE_TIME, BATTERY_DATA, CAR_INFO_DATA, ODO_METER_DATA
 
@@ -18,21 +20,16 @@ def __init__(self, username: str, password: str) -> None:
         self._client_session = httpx.AsyncClient()
 
     async def init(self):
-        await self.auth.get_token()
-
-        if self.auth.access_token is None:
-            return
+        try:
+            await self.auth.get_token()
 
-        result = await self.get_vehicle_data()
+            if self.auth.access_token is None:
+                return
 
-        # check if there are cars in the account
-        if result['data'][CAR_INFO_DATA] is None or len(result['data'][CAR_INFO_DATA]) == 0:
-            _LOGGER.exception("No cars found in account")
-            # throw new exception
-            raise Exception("No cars found in account")
+            await self._get_vehicle_data()
 
-        self.cache_data[CAR_INFO_DATA] = {
-            'data': result['data'][CAR_INFO_DATA][0], 'timestamp': datetime.now()}
+        except PolestarAuthException as e:
+            _LOGGER.exception("Auth Exception: %s", str(e))
 
     def get_latest_data(self, query: str, field_name: str) -> dict or bool or None:
         if self.cache_data and self.cache_data[query]:
@@ -61,30 +58,48 @@ def _get_field_name_value(self, field_name: str, data: dict) -> str or bool or N
             return data[field_name]
         return None
 
-    async def getOdometerData(self, vin: str):
-        result = await self.get_odo_data(vin)
+    async def _get_odometer_data(self, vin: str):
+        params = {
+            "query": "query GetOdometerData($vin: String!) { getOdometerData(vin: $vin) { averageSpeedKmPerHour eventUpdatedTimestamp { iso unix __typename } odometerMeters tripMeterAutomaticKm tripMeterManualKm __typename }}",
+            "operationName": "GetOdometerData",
+            "variables": "{\"vin\":\"" + vin + "\"}"
+        }
+        result = await self.get_graph_ql(params)
 
         if result and result['data']:
             # put result in cache
             self.cache_data[ODO_METER_DATA] = {
                 'data': result['data'][ODO_METER_DATA], 'timestamp': datetime.now()}
 
-    async def getBatteryData(self, vin: str):
-        result = await self.get_battery_data(vin)
+    async def _get_battery_data(self, vin: str):
+        params = {
+            "query": "query GetBatteryData($vin: String!) {  getBatteryData(vin: $vin) {    averageEnergyConsumptionKwhPer100Km    batteryChargeLevelPercentage    chargerConnectionStatus    chargingCurrentAmps    chargingPowerWatts    chargingStatus    estimatedChargingTimeMinutesToTargetDistance    estimatedChargingTimeToFullMinutes    estimatedDistanceToEmptyKm    estimatedDistanceToEmptyMiles    eventUpdatedTimestamp {      iso      unix      __typename    }    __typename  }}",
+            "operationName": "GetBatteryData",
+            "variables": "{\"vin\":\"" + vin + "\"}"
+        }
+
+        result = await self.get_graph_ql(params)
 
         if result and result['data']:
             # put result in cache
             self.cache_data[BATTERY_DATA] = {
                 'data': result['data'][BATTERY_DATA], 'timestamp': datetime.now()}
 
-    async def get_vehicle_data(self):
-        result = await self.get_vehicle_data()
+    async def _get_vehicle_data(self):
+        # get Vehicle Data
+        params = {
+            "query": "query getCars {  getConsumerCarsV2 {    vin    internalVehicleIdentifier    modelYear    content {      model {        code        name        __typename      }      images {        studio {          url          angles          __typename        }        __typename      }      __typename    }    hasPerformancePackage    registrationNo    deliveryDate    currentPlannedDeliveryDate    __typename  }}",
+            "operationName": "getCars",
+            "variables": "{}"
+        }
+
+        result = await self.get_graph_ql(params)
         if result and result['data']:
             # check if there are cars in the account
             if result['data'][CAR_INFO_DATA] is None or len(result['data'][CAR_INFO_DATA]) == 0:
                 _LOGGER.exception("No cars found in account")
                 # throw new exception
-                raise Exception("No cars found in account")
+                raise PolestarNoDataException("No cars found in account")
 
             self.cache_data[CAR_INFO_DATA] = {
                 'data': result['data'][CAR_INFO_DATA][0], 'timestamp': datetime.now()}
@@ -93,8 +108,20 @@ async def get_ev_data(self, vin: str):
         if self.updating is True:
             return
         self.updating = True
-        await self.getOdometerData(vin)
-        await self.getBatteryData(vin)
+        try:
+            await self._get_odometer_data(vin)
+        except PolestarNotAuthorizedException:
+            await self.auth.get_token()
+        except PolestarApiException as e:
+            _LOGGER.warning('Failed to get Odo Meter data %s', str(e))
+
+        try:
+            await self._get_battery_data(vin)
+        except PolestarNotAuthorizedException:
+            await self.auth.get_token()
+        except PolestarApiException as e:
+            _LOGGER.exception('Failed to get Battery data %s', str(e))
+
         self.updating = False
 
     def get_cache_data(self, query: str, field_name: str, skip_cache: bool = False):
@@ -123,44 +150,13 @@ async def get_graph_ql(self, params: dict):
 
         result = await self._client_session.get("https://pc-api.polestar.com/eu-north-1/my-star/", params=params, headers=headers)
         self.latest_call_code = result.status_code
+        if result.status_code == 401:
+            raise PolestarNotAuthorizedException("Unauthorized Exception")
+
+        if result.status_code != 200:
+            raise PolestarApiException(
+                f"Get GraphQL error: {result.text}")
         resultData = result.json()
 
-        # if auth error, get new token
-        if resultData.get('errors'):
-            if resultData['errors'][0]['message'] == 'User not authenticated':
-                await self.auth.get_token()
-                resultData = await self.get_graph_ql(params)
-            else:
-                # log the error
-                _LOGGER.warning(resultData.get('errors'))
-                self.latest_call_code = 500  # set internal error
-                return None
         _LOGGER.debug(resultData)
         return resultData
-
-    async def get_battery_data(self, vin: str):
-        # get Battery Data
-        params = {
-            "query": "query GetBatteryData($vin: String!) {  getBatteryData(vin: $vin) {    averageEnergyConsumptionKwhPer100Km    batteryChargeLevelPercentage    chargerConnectionStatus    chargingCurrentAmps    chargingPowerWatts    chargingStatus    estimatedChargingTimeMinutesToTargetDistance    estimatedChargingTimeToFullMinutes    estimatedDistanceToEmptyKm    estimatedDistanceToEmptyMiles    eventUpdatedTimestamp {      iso      unix      __typename    }    __typename  }}",
-            "operationName": "GetBatteryData",
-            "variables": "{\"vin\":\"" + vin + "\"}"
-        }
-        return await self.get_graph_ql(params)
-
-    async def get_vehicle_data(self):
-        # get Vehicle Data
-        params = {
-            "query": "query getCars {  getConsumerCarsV2 {    vin    internalVehicleIdentifier    modelYear    content {      model {        code        name        __typename      }      images {        studio {          url          angles          __typename        }        __typename      }      __typename    }    hasPerformancePackage    registrationNo    deliveryDate    currentPlannedDeliveryDate    __typename  }}",
-            "operationName": "getCars",
-            "variables": "{}"
-        }
-        return await self.get_graph_ql(params)
-
-    async def get_odo_data(self, vin: str):
-        # get Odo Data
-        params = {
-            "query": "query GetOdometerData($vin: String!) { getOdometerData(vin: $vin) { averageSpeedKmPerHour eventUpdatedTimestamp { iso unix __typename } odometerMeters tripMeterAutomaticKm tripMeterManualKm __typename }}",
-            "operationName": "GetOdometerData",
-            "variables": "{\"vin\":\"" + vin + "\"}"
-        }
-        return await self.get_graph_ql(params)