Tech : Chainer les appels pour .raise_for_status() d'HTTPX

itou/cities/management/commands/sync_cities.py

@@ -24,9 +24,11 @@ def fetch_cities(districts_only=False):
     }
     if districts_only:
         params["type"] = "arrondissement-municipal"
-    response = httpx.get(urllib.parse.urljoin(settings.API_GEO_BASE_URL, f"communes?{urllib.parse.urlencode(params)}"))
-    response.raise_for_status()
-    answer = response.json()
+    answer = (
+        httpx.get(urllib.parse.urljoin(settings.API_GEO_BASE_URL, f"communes?{urllib.parse.urlencode(params)}"))
+        .raise_for_status()
+        .json()
+    )
     if districts_only:
         answer = [strip_arrondissement(raw_city) for raw_city in answer]
     return answer

itou/emails/admin.py

@@ -104,10 +104,12 @@ def wrapper(*args, **kwargs):
 
     def mailjet_view(self, request, message_id, *args, **kwargs):
         # Proxy Mailjet API to avoid giving API credentials to clients.
-        response = httpx.get(
-            f"https://api.mailjet.com/v3/REST/messagehistory/{message_id}",
-            auth=(settings.ANYMAIL["MAILJET_API_KEY"], settings.ANYMAIL["MAILJET_SECRET_KEY"]),
-        )
-        response.raise_for_status()
         # Middlewares processing the response expect a Django response.
-        return JsonResponse(response.json())
+        return JsonResponse(
+            httpx.get(
+                f"https://api.mailjet.com/v3/REST/messagehistory/{message_id}",
+                auth=(settings.ANYMAIL["MAILJET_API_KEY"], settings.ANYMAIL["MAILJET_SECRET_KEY"]),
+            )
+            .raise_for_status()
+            .json()
+        )

itou/metabase/db.py

@@ -109,21 +109,19 @@ def create_table(table_name: str, columns: list[str, str], reset=False):
 def build_dbt_daily():
     # FIXME(vperron): this has to be moved to DBT seeds.
     create_unversioned_tables_if_needed()
-    response = httpx.post(
+    httpx.post(
         urllib.parse.urljoin(settings.AIRFLOW_BASE_URL, "api/v1/dags/dbt_daily/dagRuns"),
         json={"conf": {}},
-    )
-    response.raise_for_status()
+    ).raise_for_status()
 
 
 def build_dbt_weekly():
     # FIXME(vperron): this has to be moved to DBT seeds.
     create_unversioned_tables_if_needed()
-    response = httpx.post(
+    httpx.post(
         urllib.parse.urljoin(settings.AIRFLOW_BASE_URL, "api/v1/dags/dbt_weekly/dagRuns"),
         json={"conf": {}},
-    )
-    response.raise_for_status()
+    ).raise_for_status()
 
 
 def create_unversioned_tables_if_needed():

itou/metabase/management/commands/populate_metabase_matomo.py

@@ -91,9 +91,11 @@ def matomo_api_call(options):
     @tenacity.retry(stop=tenacity.stop_after_attempt(3), wait=tenacity.wait_fixed(30), after=log_retry_attempt)
     def get_csv_raw_data():
         url = urllib.parse.urljoin(settings.MATOMO_BASE_URL, "index.php")
-        response = client.get(f"{url}?{urllib.parse.urlencode(options)}", timeout=MATOMO_TIMEOUT)
-        response.raise_for_status()
-        return response.content.decode("utf-16")
+        return (
+            client.get(f"{url}?{urllib.parse.urlencode(options)}", timeout=MATOMO_TIMEOUT)
+            .raise_for_status()
+            .content.decode("utf-16")
+        )
 
     yield from csv.DictReader(io.StringIO(get_csv_raw_data()), dialect="excel")
 

itou/rdv_insertion/api.py

@@ -33,8 +33,7 @@ def get_api_credentials(refresh=False):
                         "email": settings.RDV_SOLIDARITES_EMAIL,
                         "password": settings.RDV_SOLIDARITES_PASSWORD,
                     },
-                )
-                response.raise_for_status()
+                ).raise_for_status()
                 api_credentials = {
                     "access-token": response.headers["access-token"],
                     "client": response.headers["client"],

itou/utils/apis/api_entreprise.py

@@ -26,13 +26,15 @@ class Etablissement:
 
 def get_access_token():
     try:
-        r = httpx.post(
-            f"{settings.API_INSEE_BASE_URL}/token",
-            data={"grant_type": "client_credentials"},
-            auth=(settings.API_INSEE_CONSUMER_KEY, settings.API_INSEE_CONSUMER_SECRET),
+        access_token = (
+            httpx.post(
+                f"{settings.API_INSEE_BASE_URL}/token",
+                data={"grant_type": "client_credentials"},
+                auth=(settings.API_INSEE_CONSUMER_KEY, settings.API_INSEE_CONSUMER_SECRET),
+            )
+            .raise_for_status()
+            .json()["access_token"]
         )
-        r.raise_for_status()
-        access_token = r.json()["access_token"]
     except Exception:
         logger.exception("Failed to retrieve an access token")
         return None
@@ -56,8 +58,7 @@ def etablissement_get_or_error(siret):
             url,
             headers={"Authorization": f"Bearer {access_token}"},
             params={"date": timezone.localdate().isoformat()},
-        )
-        r.raise_for_status()
+        ).raise_for_status()
     except httpx.RequestError:
         logger.exception("A request to the INSEE API failed")
         return None, "Problème de connexion à la base Sirene. Essayez ultérieurement."

itou/utils/apis/api_particulier.py

@@ -39,9 +39,7 @@ def _build_params_from(job_seeker):
 
 def _request(client, endpoint, job_seeker):
     params = _build_params_from(job_seeker=job_seeker)
-    response = client.get(endpoint, params=params)
-    response.raise_for_status()
-    return response.json()
+    return client.get(endpoint, params=params).raise_for_status().json()
 
 
 def has_required_info(job_seeker):

itou/utils/apis/data_inclusion.py

@@ -39,8 +39,7 @@ def search_services(self, code_insee: str) -> list[dict]:
                     "sources": settings.API_DATA_INCLUSION_SOURCES,
                     "thematiques": API_THEMATIQUES,
                 },
-            )
-            response.raise_for_status()
+            ).raise_for_status()
         except httpx.HTTPError as exc:
             logger.info("data.inclusion request error code_insee=%s error=%s", code_insee, exc)
             raise DataInclusionApiException()
@@ -53,12 +52,13 @@ def search_services(self, code_insee: str) -> list[dict]:
 
     def retrieve_service(self, source: str, id_: str) -> dict:
         try:
-            response = self.client.get(
-                f"/services/{source}/{id_}",
+            return (
+                self.client.get(
+                    f"/services/{source}/{id_}",
+                )
+                .raise_for_status()
+                .json()
             )
-            response.raise_for_status()
         except httpx.HTTPError as exc:
             logger.info("data.inclusion request error source=%s service_id=%s error=%s", source, id_, exc)
             raise DataInclusionApiException()
-
-        return response.json()

itou/utils/apis/datadog.py

@@ -33,9 +33,7 @@ def __init__(self):
 
     @tenacity.retry(wait=tenacity.wait_fixed(2), stop=tenacity.stop_after_attempt(8))
     def _request(self, data):
-        response = self.client.post("/logs/analytics/aggregate", content=json.dumps(data))
-        response.raise_for_status()
-        return response
+        return self.client.post("/logs/analytics/aggregate", content=json.dumps(data)).raise_for_status()
 
     def _get_data_from_datadog(self, data=None):
         response = self._request(data)

itou/utils/apis/esd.py

@@ -26,20 +26,22 @@ def get_access_token(scope):
             logger.debug("Found %s in cache. Expiration = %s, now = %s.", token.value, token.expiration, now)
             return token.value
 
-    auth_request = httpx.post(
-        f"{settings.API_ESD['AUTH_BASE_URL']}/connexion/oauth2/access_token",
-        params={"realm": "/partenaire"},
-        data={
-            "grant_type": "client_credentials",
-            "client_id": settings.API_ESD["KEY"],
-            "client_secret": settings.API_ESD["SECRET"],
-            "scope": f"application_{settings.API_ESD['KEY']} {scope}",
-        },
-        headers={"Content-Type": "application/x-www-form-urlencoded"},
+    r = (
+        httpx.post(
+            f"{settings.API_ESD['AUTH_BASE_URL']}/connexion/oauth2/access_token",
+            params={"realm": "/partenaire"},
+            data={
+                "grant_type": "client_credentials",
+                "client_id": settings.API_ESD["KEY"],
+                "client_secret": settings.API_ESD["SECRET"],
+                "scope": f"application_{settings.API_ESD['KEY']} {scope}",
+            },
+            headers={"Content-Type": "application/x-www-form-urlencoded"},
+        )
+        .raise_for_status()
+        .json()
     )
-    auth_request.raise_for_status()
 
-    r = auth_request.json()
     value = f"{r['token_type']} {r['access_token']}"
     expiration = datetime.datetime.now() + datetime.timedelta(seconds=r["expires_in"])
     token = Token(value=value, expiration=expiration)

itou/utils/apis/geiq_label.py

@@ -40,12 +40,14 @@ def __init__(self, base_url: str, token: str):
     def _command(self, command, **params):
         command = LabelCommand(command)
         try:
-            response = self.client.get(
-                f"rest/{command}",
-                params=params,
+            response_data = (
+                self.client.get(
+                    f"rest/{command}",
+                    params=params,
+                )
+                .raise_for_status()
+                .json()
             )
-            response.raise_for_status()
-            response_data = response.json()
         except (httpx.HTTPError, json.JSONDecodeError) as exc:
             raise LabelAPIError("Error requesting Label API") from exc
         if response_data.get("status") != "Success":

itou/utils/apis/geocoding.py

@@ -46,8 +46,7 @@ def call_ban_geocoding_api(address, post_code=None, limit=1):
     url = f"{api_url}?{query_string}"
 
     try:
-        r = httpx.get(url)
-        r.raise_for_status()
+        r = httpx.get(url).raise_for_status()
     except httpx.HTTPError as e:
         logger.info("Error while requesting `%s`: %s", url, e)
         return None

itou/utils/apis/pole_emploi.py

@@ -89,19 +89,21 @@ def __init__(self, base_url, auth_base_url, key, secret):
 
     def _refresh_token(self):
         scopes = " ".join(AUTHORIZED_SCOPES)
-        response = httpx.post(
-            f"{self.auth_base_url}/connexion/oauth2/access_token",
-            params={"realm": "/partenaire"},
-            data={
-                "client_id": self.key,
-                "client_secret": self.secret,
-                "grant_type": "client_credentials",
-                "scope": f"application_{self.key} {scopes}",
-            },
-            headers={"Content-Type": "application/x-www-form-urlencoded"},
+        auth_data = (
+            httpx.post(
+                f"{self.auth_base_url}/connexion/oauth2/access_token",
+                params={"realm": "/partenaire"},
+                data={
+                    "client_id": self.key,
+                    "client_secret": self.secret,
+                    "grant_type": "client_credentials",
+                    "scope": f"application_{self.key} {scopes}",
+                },
+                headers={"Content-Type": "application/x-www-form-urlencoded"},
+            )
+            .raise_for_status()
+            .json()
         )
-        response.raise_for_status()
-        auth_data = response.json()
         token = f"{auth_data['token_type']} {auth_data['access_token']}"
         caches["failsafe"].set(
             CACHE_API_TOKEN_KEY,

itou/utils/apis/sentry.py

@@ -28,9 +28,7 @@ def _request(self, start, end):
             "end": end.isoformat(),
         }
 
-        response = self.client.get("/events/", params=params)
-        response.raise_for_status()
-        return response
+        return self.client.get("/events/", params=params).raise_for_status()
 
     def get_metrics(self, start, end):
         response = self._request(start=start, end=end)

itou/utils/apis/updown.py

@@ -24,9 +24,7 @@ def _request(self, endpoint, start, end):
             "from": start.isoformat(),
             "to": end.isoformat(),
         }
-        response = self.client.get(endpoint, params=params)
-        response.raise_for_status()
-        return response
+        return self.client.get(endpoint, params=params).raise_for_status()
 
     def get_metrics(self, start, end):
         endpoint = f"/checks/{settings.API_UPDOWN_CHECK_ID}/metrics/"

itou/www/apply/views/process_views.py

@@ -1215,8 +1215,7 @@ def rdv_insertion_invite(request, job_application_id, for_detail=False):
                 if response.status_code in (httpx.codes.UNAUTHORIZED, httpx.codes.FORBIDDEN):
                     headers = get_api_credentials(refresh=True)
                     response = httpx.post(url=url, headers=headers, json=data, timeout=10)
-                response.raise_for_status()
-                response_data = response.json()
+                response_data = response.raise_for_status().json()
 
                 invitation_request = InvitationRequest.objects.create(
                     job_seeker=job_application.job_seeker,