chore: remove unnecessary checks and cleanup error handling (#284)

- properly raise 401 errors in SIRENE calls
- more consistent handling of errors and 404 in `get_sirene_relatives`
- remove config file check in `_get_credentials_from_configfile` (checks are done in PynseeAPISession._request_api_insee)
- cleanup error handling in PynseeAPISession._request_api_insee

Author: tfardet

pynsee/sirene/_request_sirene.py

@@ -38,7 +38,7 @@ def _request_sirene(query, kind, number=1001):
     if number > number_query_limit:
         link = link + "&curseur=*"
 
-    with PynseeAPISession(url=link) as session:
+    with PynseeAPISession() as session:
         request = session.request_insee(
             api_url=link,
             file_format="application/json;charset=utf-8",
@@ -84,7 +84,7 @@ def _request_sirene(query, kind, number=1001):
                     + following_cursor
                 )
 
-                with PynseeAPISession(url=new_query) as session:
+                with PynseeAPISession() as session:
                     request_new = session.request_insee(
                         api_url=new_query,
                         file_format="application/json;charset=utf-8",

pynsee/sirene/get_sirene_data.py

@@ -1,18 +1,19 @@
 # -*- coding: utf-8 -*-
 # Copyright : INSEE, 2021
 
-import pandas as pd
-from functools import lru_cache
+import logging
 import re
+from functools import lru_cache
+
+import pandas as pd
+from requests import RequestException
 
 from pynsee.utils.requests_session import PynseeAPISession
 from pynsee.utils._make_dataframe_from_dict import _make_dataframe_from_dict
 from pynsee.utils.HiddenPrints import HiddenPrints
 
 from .sirenedataframe import SireneDataFrame
 
-import logging
-
 
 logger = logging.getLogger(__name__)
 
@@ -51,21 +52,19 @@ def get_sirene_data(*id):
 
     list_data = []
 
-    for i in range(len(list_ids)):
+    for sid in list_ids:
         for kind in ["siret", "siren"]:
             if kind == "siren":
                 main_key = "uniteLegale"
             elif kind == "siret":
                 main_key = "etablissement"
 
             INSEE_api_sirene = "https://api.insee.fr/api-sirene/3.11/" + kind
-            link = (
-                INSEE_api_sirene + "/" + re.sub(r"\s+", "", str(list_ids[i]))
-            )
+            link = INSEE_api_sirene + "/" + re.sub(r"\s+", "", str(sid))
 
             try:
                 with HiddenPrints():
-                    with PynseeAPISession(url=link) as session:
+                    with PynseeAPISession() as session:
                         request = session.request_insee(
                             api_url=link,
                             file_format="application/json;charset=utf-8",
@@ -85,19 +84,20 @@ def get_sirene_data(*id):
                     data = data_request[main_key]
 
                 data_final = _make_dataframe_from_dict(data)
+            except RequestException as e:
+                if e.response.status_code == 401:
+                    raise
             except Exception:
                 pass
             else:
                 list_data.append(data_final)
                 break
 
-    if len(list_data) > 0:
+    if list_data:
         data_final = pd.concat(list_data).reset_index(drop=True)
-    else:
-        raise ValueError("!!! No data found for the provided identifiers !!!")
 
-    _warning_get_data()
+        _warning_get_data()
 
-    SireneDF = SireneDataFrame(data_final)
+        return SireneDataFrame(data_final)
 
-    return SireneDF
+    raise ValueError("!!! No data found for the provided identifiers !!!")

pynsee/sirene/get_sirene_relatives.py

@@ -1,9 +1,10 @@
 import pandas as pd
 import re
 
+from requests import RequestException
+
 from pynsee.utils.requests_session import PynseeAPISession
 from pynsee.utils._make_dataframe_from_dict import _make_dataframe_from_dict
-from pynsee.utils.HiddenPrints import HiddenPrints
 from .sirenedataframe import SireneDataFrame
 
 
@@ -24,7 +25,6 @@ def get_sirene_relatives(*siret):
         >>> data = get_sirene_relatives('00555008200027')
         >>> data = get_sirene_relatives(['39860733300059', '00555008200027'])
     """
-
     list_siret = []
 
     for id in range(len(siret)):
@@ -40,38 +40,44 @@ def get_sirene_relatives(*siret):
     types = ["siretEtablissementPredecesseur", "siretEtablissementSuccesseur"]
     list_df = []
 
-    for s in range(len(list_siret)):
-        for i in range(len(types)):
-
-            criteria = types[i] + ":" + re.sub(r"\s+", "", list_siret[s])
+    for s in list_siret:
+        for t in types:
+            criteria = t + ":" + re.sub(r"\s+", "", s)
             query = (
                 "https://api.insee.fr/api-sirene/3.11/siret/liensSuccession"
                 f"?q={criteria}"
             )
+
             try:
-                with HiddenPrints():
-                    with PynseeAPISession(url=query) as session:
-                        result = session.request_insee(
-                            api_url=query,
-                            file_format="application/json;charset=utf-8",
-                            raise_if_not_ok=True,
-                            print_msg=False,
-                        )
-
-                    json = result.json()
+                with PynseeAPISession() as session:
+                    result = session.request_insee(
+                        api_url=query,
+                        file_format="application/json;charset=utf-8",
+                        raise_if_not_ok=True,
+                        print_msg=False,
+                    )
+
+                json = result.json()
+            except RequestException as e:
+                if e.response.status_code == 401:
+                    raise
             except Exception:
                 pass
             else:
                 list_df += [_make_dataframe_from_dict(json)]
 
-    if len(list_df) > 0:
-        df = SireneDataFrame(pd.concat(list_df).reset_index(drop=True))
-
-        for c in ["statut", "message", "nombre", "total", "debut"]:
-            if c in df.columns:
-                del df[c]
+    if list_df:
+        df = SireneDataFrame(
+            pd.concat(list_df)
+            .drop(
+                columns=["statut", "message", "nombre", "total", "debut"],
+                errors="ignore",
+            )
+            .reset_index(drop=True)
+        )
 
-        return df
+        if df.columns.any():
+            return df
 
     raise ValueError(
         "Neither parent nor child entities were found for any entity"

pynsee/utils/_get_credentials.py

@@ -3,44 +3,25 @@
 
 import json
 import logging
-import re
-
-from functools import lru_cache
-from typing import Dict
 
 from pynsee.constants import CONFIG_FILE
 
 
 logger = logging.getLogger(__name__)
 
 
-def _get_credentials_from_configfile(url) -> Dict[str, str]:
+def _get_credentials_from_configfile() -> dict[str, str]:
     """
     Try to load credentials and proxy configuration from config file.
 
     Returns a dict containing at least an "sirene_key" entry.
     """
-    key_dict: Dict[str, str] = {}
+    key_dict: dict[str, str] = {}
 
     try:
         with open(CONFIG_FILE, "r") as f:
             key_dict = json.load(f)
-
     except FileNotFoundError:
-        if re.match(".*api-sirene.*", url):
-            # no credentials/config stored
-            _missing_credentials()
-        return key_dict
+        pass
 
     return key_dict
-
-
-@lru_cache(maxsize=None)
-def _missing_credentials() -> None:
-    logger.critical(
-        "INSEE API credentials have not been found: please try to reuse "
-        "pynsee.init_conn to save them locally.\n"
-        "Otherwise, you can still use environment variables as follow:\n\n"
-        "import os\n"
-        "os.environ['sirene_key'] = 'my_sirene_key'"
-    )