[#84] Add a function to retrieve list of PCE ids and labels from account

Author: Stéphane Senart

pygazpar/api_client.py

@@ -1,10 +1,13 @@
-from requests import Session, Response
-from typing import Any
 import http.cookiejar
 import json
+import logging
+import time
+import traceback
 from datetime import date
 from enum import Enum
-import time
+from typing import Any
+
+from requests import Response, Session
 
 SESSION_TOKEN_URL = "https://connexion.grdf.fr/api/v1/authn"
 SESSION_TOKEN_PAYLOAD = """{{
@@ -27,6 +30,8 @@
 
 DATE_FORMAT = "%Y-%m-%d"
 
+Logger = logging.getLogger(__name__)
+
 
 # ------------------------------------------------------
 class ConsumptionType(str, Enum):
@@ -36,11 +41,11 @@ class ConsumptionType(str, Enum):
 
 # ------------------------------------------------------
 class Frequency(str, Enum):
-    HOURLY = 'Horaire'
-    DAILY = 'Journalier'
-    WEEKLY = 'Hebdomadaire'
-    MONTHLY = 'Mensuel'
-    YEARLY = 'Annuel'
+    HOURLY = "Horaire"
+    DAILY = "Journalier"
+    WEEKLY = "Hebdomadaire"
+    MONTHLY = "Mensuel"
+    YEARLY = "Annuel"
 
 
 # ------------------------------------------------------
@@ -114,7 +119,9 @@ def get(self, endpoint: str, params: dict[str, Any]) -> Response:
                 response = self._session.get(f"{API_BASE_URL}{endpoint}", params=params)
 
                 if "text/html" in response.headers.get("Content-Type"):  # type: ignore
-                    raise ValueError(f"An error occurred while loading data. Please check your query parameters: {params}")
+                    raise ValueError(
+                        f"An error occurred while loading data. Please check your query parameters: {params}"
+                    )
 
                 if response.status_code != 200:
                     raise ValueError(
@@ -124,7 +131,9 @@ def get(self, endpoint: str, params: dict[str, Any]) -> Response:
                 break
             except Exception as e:  # pylint: disable=broad-exception-caught
                 if retry == 1:
+                    Logger.error(f"An error occurred while loading data. Retry limit reached: {traceback.format_exc()}")
                     raise e
+                Logger.warning("An error occurred while loading data. Retry in 3 seconds...")
                 time.sleep(3)
                 retry -= 1
 
@@ -141,14 +150,16 @@ def get_pce_list(self, details: bool = False) -> list[Any]:
         return res
 
     # ------------------------------------------------------
-    def get_pce_consumption(self, consumption_type: ConsumptionType, start_date: date, end_date: date, pce_list: list[str]) -> dict[str, Any]:
+    def get_pce_consumption(
+        self, consumption_type: ConsumptionType, start_date: date, end_date: date, pce_list: list[str]
+    ) -> dict[str, Any]:
 
         start = start_date.strftime(DATE_FORMAT)
         end = end_date.strftime(DATE_FORMAT)
 
         res = self.get(
             f"/e-conso/pce/consommation/{consumption_type.value}",
-            {"dateDebut": start, "dateFin": end, "pceList[]": ",".join(pce_list)}
+            {"dateDebut": start, "dateFin": end, "pceList[]": ",".join(pce_list)},
         ).json()
 
         if type(res) is list and len(res) == 0:
@@ -160,14 +171,21 @@ def get_pce_consumption(self, consumption_type: ConsumptionType, start_date: dat
         return res
 
     # ------------------------------------------------------
-    def get_pce_consumption_excelsheet(self, consumption_type: ConsumptionType, start_date: date, end_date: date, frequency: Frequency, pce_list: list[str]) -> dict[str, Any]:
+    def get_pce_consumption_excelsheet(
+        self,
+        consumption_type: ConsumptionType,
+        start_date: date,
+        end_date: date,
+        frequency: Frequency,
+        pce_list: list[str],
+    ) -> dict[str, Any]:
 
         start = start_date.strftime(DATE_FORMAT)
         end = end_date.strftime(DATE_FORMAT)
 
         response = self.get(
             f"/e-conso/pce/consommation/{consumption_type.value}/telecharger",
-            {"dateDebut": start, "dateFin": end, "frequence": frequency.value, "pceList[]": ",".join(pce_list)}
+            {"dateDebut": start, "dateFin": end, "frequence": frequency.value, "pceList[]": ",".join(pce_list)},
         )
 
         filename = response.headers["Content-Disposition"].split("filename=")[1]
@@ -181,10 +199,7 @@ def get_pce_meteo(self, end_date: date, days: int, pce: str) -> dict[str, Any]:
 
         end = end_date.strftime(DATE_FORMAT)
 
-        res = self.get(
-            f"/e-conso/pce/{pce}/meteo",
-            {"dateFinPeriode": end, "nbJours": days}
-        ).json()
+        res = self.get(f"/e-conso/pce/{pce}/meteo", {"dateFinPeriode": end, "nbJours": days}).json()
 
         if type(res) is list and len(res) == 0:
             return dict[str, Any]()

pygazpar/datasource.py

@@ -1,37 +1,18 @@
 import glob
-import http.cookiejar
 import json
 import logging
 import os
-import time
 from abc import ABC, abstractmethod
 from datetime import date, timedelta
 from typing import Any, Dict, List, Optional, cast
 
 import pandas as pd
-from requests import Session
 
+from pygazpar.api_client import APIClient, ConsumptionType
+from pygazpar.api_client import Frequency as APIClientFrequency
 from pygazpar.enum import Frequency, PropertyName
 from pygazpar.excelparser import ExcelParser
 from pygazpar.jsonparser import JsonParser
-from pygazpar.api_client import APIClient, ConsumptionType, Frequency as APIClientFrequency
-
-SESSION_TOKEN_URL = "https://connexion.grdf.fr/api/v1/authn"
-SESSION_TOKEN_PAYLOAD = """{{
-    "username": "{0}",
-    "password": "{1}",
-    "options": {{
-        "multiOptionalFactorEnroll": "false",
-        "warnBeforePasswordExpired": "false"
-    }}
-}}"""
-
-AUTH_TOKEN_URL = "https://connexion.grdf.fr/login/sessionCookieRedirect"
-AUTH_TOKEN_PARAMS = """{{
-    "checkAccountSetupComplete": "true",
-    "token": "{0}",
-    "redirectUrl": "https://monespace.grdf.fr"
-}}"""
 
 Logger = logging.getLogger(__name__)
 
@@ -75,46 +56,6 @@ def load(
 
         return res
 
-    # ------------------------------------------------------
-    def _login(self, username: str, password: str) -> str:
-
-        session = Session()
-        session.headers.update({"domain": "grdf.fr"})
-        session.headers.update({"Content-Type": "application/json"})
-        session.headers.update({"X-Requested-With": "XMLHttpRequest"})
-
-        payload = SESSION_TOKEN_PAYLOAD.format(username, password)
-
-        response = session.post(SESSION_TOKEN_URL, data=payload)
-
-        if response.status_code != 200:
-            raise ValueError(
-                f"An error occurred while logging in. Status code: {response.status_code} - {response.text}"
-            )
-
-        session_token = response.json().get("sessionToken")
-
-        Logger.debug("Session token: %s", session_token)
-
-        jar = http.cookiejar.CookieJar()
-
-        self._session = Session()  # pylint: disable=attribute-defined-outside-init
-        self._session.headers.update({"Content-Type": "application/json"})
-        self._session.headers.update({"X-Requested-With": "XMLHttpRequest"})
-
-        params = json.loads(AUTH_TOKEN_PARAMS.format(session_token))
-
-        response = self._session.get(AUTH_TOKEN_URL, params=params, allow_redirects=True, cookies=jar)  # type: ignore
-
-        if response.status_code != 200:
-            raise ValueError(
-                f"An error occurred while getting the auth token. Status code: {response.status_code} - {response.text}"
-            )
-
-        auth_token = self._session.cookies.get("auth_token", domain="monespace.grdf.fr")
-
-        return auth_token  # type: ignore
-
     @abstractmethod
     def _loadFromSession(
         self, pceIdentifier: str, startDate: date, endDate: date, frequencies: Optional[List[Frequency]] = None
@@ -125,8 +66,6 @@ def _loadFromSession(
 # ------------------------------------------------------------------------------------------------------------
 class ExcelWebDataSource(WebDataSource):  # pylint: disable=too-few-public-methods
 
-    DATA_URL = "https://monespace.grdf.fr/api/e-conso/pce/consommation/informatives/telecharger?dateDebut={0}&dateFin={1}&frequence={3}&pceList[]={2}"
-
     DATE_FORMAT = "%Y-%m-%d"
 
     FREQUENCY_VALUES = {
@@ -173,19 +112,18 @@ def _loadFromSession(  # pylint: disable=too-many-branches
             frequencyList = list(set(frequencies))
 
         for frequency in frequencyList:
-            # Inject parameters.
-            downloadUrl = ExcelWebDataSource.DATA_URL.format(
-                startDate.strftime(ExcelWebDataSource.DATE_FORMAT),
-                endDate.strftime(ExcelWebDataSource.DATE_FORMAT),
-                pceIdentifier,
-                ExcelWebDataSource.FREQUENCY_VALUES[frequency],
-            )
 
             Logger.debug(
                 f"Loading data of frequency {ExcelWebDataSource.FREQUENCY_VALUES[frequency]} from {startDate.strftime(ExcelWebDataSource.DATE_FORMAT)} to {endDate.strftime(ExcelWebDataSource.DATE_FORMAT)}"
             )
 
-            response = self._api_client.get_pce_consumption_excelsheet(ConsumptionType.INFORMATIVE, startDate, endDate, APIClientFrequency(ExcelWebDataSource.FREQUENCY_VALUES[frequency]), [pceIdentifier])
+            response = self._api_client.get_pce_consumption_excelsheet(
+                ConsumptionType.INFORMATIVE,
+                startDate,
+                endDate,
+                APIClientFrequency(ExcelWebDataSource.FREQUENCY_VALUES[frequency]),
+                [pceIdentifier],
+            )
 
             filename = response["filename"]
             content = response["content"]
@@ -249,12 +187,6 @@ def load(
 # ------------------------------------------------------------------------------------------------------------
 class JsonWebDataSource(WebDataSource):  # pylint: disable=too-few-public-methods
 
-    DATA_URL = (
-        "https://monespace.grdf.fr/api/e-conso/pce/consommation/informatives?dateDebut={0}&dateFin={1}&pceList[]={2}"
-    )
-
-    TEMPERATURES_URL = "https://monespace.grdf.fr/api/e-conso/pce/{0}/meteo?dateFinPeriode={1}&nbJours={2}"
-
     INPUT_DATE_FORMAT = "%Y-%m-%d"
 
     OUTPUT_DATE_FORMAT = "%d/%m/%Y"

tests/test_api_client.py

@@ -1,8 +1,10 @@
-from pygazpar.api_client import APIClient, ConsumptionType, Frequency
 import os
 from datetime import date
+
 import pytest
 
+from pygazpar.api_client import APIClient, ConsumptionType, Frequency
+
 
 class TestAPIClient:
 
@@ -16,7 +18,7 @@ def setup_class(cls):
         cls._password = os.environ["GRDF_PASSWORD"]
         cls._pceIdentifier = os.environ["PCE_IDENTIFIER"]
 
-        cls._client = APIClient(cls._username, cls._password)
+        cls._client = APIClient(cls._username, cls._password, 3)
         cls._client.login()
 
     # ------------------------------------------------------
@@ -74,30 +76,42 @@ def test_get_pce_list(self):
     # ------------------------------------------------------
     def test_get_pce_consumption(self):
 
-        start_date = date.today()
-        end_date = date.today()
+        start_date = date(2025, 1, 1)
+        end_date = date(2025, 1, 7)
 
-        pce_consumption_informative = TestAPIClient._client.get_pce_consumption(ConsumptionType.INFORMATIVE, start_date, end_date, [TestAPIClient._pceIdentifier])
+        pce_consumption_informative = TestAPIClient._client.get_pce_consumption(
+            ConsumptionType.INFORMATIVE, start_date, end_date, [TestAPIClient._pceIdentifier]
+        )
 
         assert len(pce_consumption_informative) > 0
 
-        pce_consumption_published = TestAPIClient._client.get_pce_consumption(ConsumptionType.PUBLISHED, start_date, end_date, [TestAPIClient._pceIdentifier])
+        pce_consumption_published = TestAPIClient._client.get_pce_consumption(
+            ConsumptionType.PUBLISHED, start_date, end_date, [TestAPIClient._pceIdentifier]
+        )
 
         assert len(pce_consumption_published) > 0
 
-        with pytest.raises(ValueError, match="An error occurred while loading data."):
-            TestAPIClient._client.get_pce_consumption(ConsumptionType.INFORMATIVE, date(2010, 1, 1), date(2010, 1, 2), [TestAPIClient._pceIdentifier])
+        no_result = TestAPIClient._client.get_pce_consumption(
+            ConsumptionType.INFORMATIVE, date(2010, 1, 1), date(2010, 1, 7), [TestAPIClient._pceIdentifier]
+        )
+
+        assert len(no_result) == 0
+
+        no_result = TestAPIClient._client.get_pce_consumption(
+            ConsumptionType.INFORMATIVE, start_date, end_date, ["InvalidPceIdentifier"]
+        )
 
-        with pytest.raises(ValueError, match="An error occurred while loading data."):
-            TestAPIClient._client.get_pce_consumption(ConsumptionType.INFORMATIVE, start_date, end_date, ["InvalidPceIdentifier"])
+        assert len(no_result) == 0
 
     # ------------------------------------------------------
     def test_get_pce_consumption_excelsheet(self):
 
-        start_date = date.today()
-        end_date = date.today()
+        start_date = date(2025, 1, 1)
+        end_date = date(2025, 1, 7)
 
-        pce_consumption_informative = TestAPIClient._client.get_pce_consumption_excelsheet(ConsumptionType.INFORMATIVE, start_date, end_date, Frequency.DAILY, [TestAPIClient._pceIdentifier])
+        pce_consumption_informative = TestAPIClient._client.get_pce_consumption_excelsheet(
+            ConsumptionType.INFORMATIVE, start_date, end_date, Frequency.DAILY, [TestAPIClient._pceIdentifier]
+        )
 
         assert len(pce_consumption_informative) > 0
 
@@ -110,9 +124,9 @@ def test_get_pce_meteo(self):
 
         assert len(pce_meteo) > 0
 
-        pce_meteo_no_result = TestAPIClient._client.get_pce_meteo(date(2010, 1, 2), 2, TestAPIClient._pceIdentifier)
+        pce_meteo_no_result = TestAPIClient._client.get_pce_meteo(date(2010, 1, 2), 7, TestAPIClient._pceIdentifier)
 
         assert len(pce_meteo_no_result) == 0
 
         with pytest.raises(ValueError, match="Le pce InvalidPceIdentifier n'existe pas !"):
-            TestAPIClient._client.get_pce_meteo(end_date, 2, "InvalidPceIdentifier")
+            TestAPIClient._client.get_pce_meteo(end_date, 7, "InvalidPceIdentifier")