use refresh token

pypolestar · Dec 31, 2023 · 0dc179e · 0dc179e
1 parent 1440d1f
commit 0dc179e
Show file tree

Hide file tree

Showing 2 changed files with 32 additions and 22 deletions.
diff --git a/custom_components/polestar_api/pypolestar/auth.py b/custom_components/polestar_api/pypolestar/auth.py
@@ -22,34 +22,44 @@ def __init__(self, username: str, password: str) -> None:
         self.latest_call_code = None
         self._client_session = httpx.AsyncClient()
 
-    async def get_token(self) -> None:
-        code = await self._get_code()
-        if code is None:
-            return
-
-        # get token
-        params = {
-            "query": "query getAuthToken($code: String!) { getAuthToken(code: $code) { id_token access_token refresh_token expires_in }}",
-            "operationName": "getAuthToken",
-            "variables": json.dumps({"code": code})
-        }
-
-        headers = {
-            "Content-Type": "application/json"
-        }
+    async def get_token(self, refresh=False) -> None:
+        # get access / refresh token
+        headers = {"Content-Type": "application/json"}
+        operationName = "getAuthToken"
+        if not refresh:
+            code = await self._get_code()
+            if code is None:
+                return
+            params = {
+                "query": "query getAuthToken($code: String!) { getAuthToken(code: $code) { id_token access_token refresh_token expires_in }}",
+                "operationName": operationName,
+                "variables": json.dumps({"code": code}),
+            }
+        else:
+            if self.refresh_token is None:
+                return
+            token = self.refresh_token
+            operationName = "refreshAuthToken"
+            headers["Authorization"] = f"Bearer {self.access_token}"
+
+            params = {
+                "query": "query refreshAuthToken($token: String!) { refreshAuthToken(token: $token) { id_token access_token refresh_token expires_in }}",
+                "operationName": operationName,
+                "variables": json.dumps({"token": token}),
+            }
         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:
+        resultData = result.json()
+        if result.status_code != 200 or ("errors" in resultData and len(resultData["errors"])):
             raise PolestarAuthException(
                 f"Error getting token", result.status_code)
-        resultData = result.json()
         _LOGGER.debug(resultData)
 
         if resultData['data']:
-            self.access_token = resultData['data']['getAuthToken']['access_token']
-            self.refresh_token = resultData['data']['getAuthToken']['refresh_token']
+            self.access_token = resultData['data'][operationName]['access_token']
+            self.refresh_token = resultData['data'][operationName]['refresh_token']
             self.token_expiry = datetime.now(
-            ) + timedelta(seconds=resultData['data']['getAuthToken']['expires_in'])
+            ) + timedelta(seconds=resultData['data'][operationName]['expires_in'])
             # ID Token
 
         _LOGGER.debug(f"Response {self.access_token}")

diff --git a/custom_components/polestar_api/pypolestar/polestar.py b/custom_components/polestar_api/pypolestar/polestar.py
@@ -111,8 +111,8 @@ async def get_ev_data(self, vin: str):
         try:
             if self.auth.token_expiry is None:
                 raise PolestarAuthException("No token expiry found")
-            if self.auth.token_expiry < datetime.now():
-                await self.auth.get_token()
+            if (self.auth.token_expiry - datetime.now()).total_seconds() < 300:
+                await self.auth.get_token(refresh=True)
         except PolestarAuthException as e:
             self.latest_call_code = 500
             _LOGGER.warning("Auth Exception: %s", str(e))