add auto token refresh on 401

doc/source/usage.rst

@@ -34,7 +34,7 @@ For the first authentication, in order to obtain an access_token and a refresh_t
 This portal will then redirect to a specific url and provide the code and state parameters.
 
 By default in the function :meth:`edf_api.auth.EDFAuth.get_login_url` this url is ``edfetmoiauth:/oauthcallback`` which will raise an error in the javascript console (because the scheme is unknown).
-You can then retrieve this error and parse it (there is an example to do so in the cli.py demo file), this method allows to retrieve the data even without a server ready to receive the data from the redirection.
+You can then retrieve this error and parse it (there is an example to do so in the cli.py demo file), this is currently the only way to retrieve the result as the redirect url is restricted (any other one will be refused by the server).
 
 .. image:: _static/error.png
    :width: 600

edf_api/__init__.py

@@ -27,14 +27,17 @@ def __init__(self, session: aiohttp.ClientSession, access_token: str, refresh_to
         self._refresh_token = refresh_token
         self._expiration = expiration
 
-    async def _get_access_token(self) -> str:
+    async def _get_access_token(self, force=False) -> str:
         """
         Function to get and access_token and ensure that it is valid
 
+        Args:
+            force: set to true to force the token refresh
+
         Returns:
             a valid access_token
         """
-        if self._expiration-5 < time.time():
+        if self._expiration-5 < time.time() or force:
             auth = EDFAuth(self._session)
             self._access_token, self._refresh_token, self._expiration = await auth.get_token(refresh_token=self._refresh_token)
         return self._access_token
@@ -69,7 +72,7 @@ async def get_info(self, insee: str, pdl: str) -> ElecInfo:
                 return ElecInfo.from_json(req["listeCoupuresInfoReseau"][i], req["listeCrises"][i])
         return ElecInfo.no_outage()
 
-    async def get_data(self, bp_num: str, pdl: str, start: str, end: str) -> dict:
+    async def get_data(self, bp_num: str, pdl: str, start: str, end: str, retry: bool = False) -> dict:
         """
         Function to get the consumption data
 
@@ -78,8 +81,9 @@ async def get_data(self, bp_num: str, pdl: str, start: str, end: str) -> dict:
             pdl: the PDL id
             start: the start date for the retrieved data in YYYY-MM-DD format
             end: the end date for the retrieved data in YYYY-MM-DD format
+            retry: (internal) used to know if this is the first call to this function or if this is a retry
         """
-        tok = await self._get_access_token()
+        tok = await self._get_access_token(retry)
         async with async_timeout.timeout(TIMEOUT):
             response = await self._session.get(f"https://api-edf.edelia.fr/api/v1/sites/-/load-curve?step=30&begin-date={start}&end-date={end}&withCost=true",
                 headers={
@@ -91,4 +95,9 @@ async def get_data(self, bp_num: str, pdl: str, start: str, end: str) -> dict:
                 },
             )
             req = await response.json()
+
+            if (("errorCode" in req and req["errorCode"] == "401") or ("status" in req and req["status"] == 401)) and not retry:
+                # try to force the access_token refresh if
+                return await self.get_data(bp_num, pdl, start, end, True)
+
         return req

setup.py

@@ -6,7 +6,7 @@
 setup(
     name='edf_api',
     packages=find_packages(include=['edf_api']),
-    version='0.1.0',
+    version='0.1.1',
     author='drosocode',
     license='MIT',
     description='API for EDF',