ants/api: refactor call to the doublon API (#83558)

src/ants_hub/api/ants.py

@@ -30,6 +30,70 @@ def get_api_optimisation_url():
 IDENTIFIANT_PREDEMANDE_RE = re.compile(r'^[A-Z0-9]{10}$')
 
 
+class APIDoublon:
+    def __init__(self):
+        self.auth_token = get_api_optimisation_auth_token()
+        if self.auth_token is None:
+            raise ANTSError('REQUEST_TO_ANTS_V2_AUTH_TOKEN not configured')
+        self.base_url = get_api_optimisation_url()
+
+    def request(self, method, endpoint, **kwargs):
+        try:
+            response = requests.request(
+                method,
+                self.base_url + endpoint,
+                headers={'x-rdv-opt-auth-token': self.auth_token},
+                timeout=10,
+                **kwargs,
+            )
+            response.raise_for_status()
+            return response.json()
+        except requests.HTTPError as e:
+            if e.response.status_code == 422:
+                raise ANTSError(str(e), e.response.json())
+            raise ANTSError(
+                str(e),
+            )
+        except (requests.RequestException, ValueError, KeyError, TypeError) as e:
+            raise ANTSError(repr(e))
+
+    def create_rdv(self, rdv):
+        identifiant_predemande = rdv.identifiant_predemande.upper()
+        params = [
+            ('application_id', identifiant_predemande),
+            ('management_url', rdv.gestion_url),
+            ('meeting_point', rdv.lieu.nom),
+            ('meeting_point_id', str(rdv.lieu.id)),
+            ('appointment_date', localtime(rdv.date).strftime('%Y-%m-%d %H:%M:%S')),
+        ]
+        response = self.request(method='POST', endpoint='appointments', params=params)
+        try:
+            return response['success']
+        except Exception as e:
+            raise ANTSError(str(e), {'response': response})
+
+    def delete_rdv(self, rdv):
+        identifiant_predemande = rdv.identifiant_predemande.upper()
+        params = [
+            ('application_id', identifiant_predemande),
+            ('meeting_point', rdv.lieu.nom),
+            ('meeting_point_id', str(rdv.lieu.id)),
+            ('appointment_date', localtime(rdv.date).strftime('%Y-%m-%d %H:%M:%S')),
+        ]
+        response = self.request(method='DELETE', endpoint='appointments', params=params)
+        try:
+            return 'rowcount' in response
+        except Exception as e:
+            raise ANTSError(str(e), {'response': response})
+
+    def statuses(self, application_ids):
+        params = [('application_ids', application_id) for application_id in application_ids]
+        data = self.request(method='GET', endpoint='status', params=params)
+        if not isinstance(data, dict):
+            raise ANTSError('response is not a dict', data)
+        return data
+
+
 def push_rdv(rdv):
     identifiant_predemande = rdv.identifiant_predemande.upper()
     if not IDENTIFIANT_PREDEMANDE_RE.match(identifiant_predemande):
@@ -38,51 +102,11 @@ def push_rdv(rdv):
         )
         return True
 
-    # swagger : https://api-coordination.rendezvouspasseport.ants.gouv.fr/docs#/Appointments%20-%20%C3%A9diteurs/add_appointment_api_appointments_post
-    auth_token = get_api_optimisation_auth_token()
-    if auth_token is None:
-        raise ANTSError('REQUEST_TO_ANTS_V2_AUTH_TOKEN not configured')
-    base_url = get_api_optimisation_url()
-    try:
-        if rdv.canceled is None:
-            params = [
-                ('application_id', identifiant_predemande),
-                ('management_url', rdv.gestion_url),
-                ('meeting_point', rdv.lieu.nom),
-                ('meeting_point_id', str(rdv.lieu.id)),
-                ('appointment_date', localtime(rdv.date).strftime('%Y-%m-%d %H:%M:%S')),
-            ]
-            response = requests.post(
-                base_url + 'appointments',
-                params=params,
-                headers={'x-rdv-opt-auth-token': auth_token},
-                timeout=10,
-            )
-            response.raise_for_status()
-            return response.json()['success']
-        else:
-            params = [
-                ('application_id', identifiant_predemande),
-                ('meeting_point', rdv.lieu.nom),
-                ('meeting_point_id', str(rdv.lieu.id)),
-                ('appointment_date', localtime(rdv.date).strftime('%Y-%m-%d %H:%M:%S')),
-            ]
-            response = requests.delete(
-                base_url + 'appointments',
-                params=params,
-                headers={'x-rdv-opt-auth-token': auth_token},
-                timeout=10,
-            )
-            response.raise_for_status()
-            return 'rowcount' in response.json()
-    except requests.HTTPError as e:
-        if e.response.status_code == 422:
-            raise ANTSError(str(e), e.response.json())
-        raise ANTSError(
-            str(e),
-        )
-    except (requests.RequestException, ValueError, KeyError, TypeError) as e:
-        raise ANTSError(str(e))
+    api_doublon = APIDoublon()
+    if rdv.canceled is None:
+        return api_doublon.create_rdv(rdv)
+    else:
+        return api_doublon.delete_rdv(rdv)
 
 
 def get_rdvs_to_upload():
@@ -110,38 +134,21 @@ def upload_rdvs():
 
 
 def get_status_of_predemandes(identifiant_predemandes: list):
-    auth_token = get_api_optimisation_auth_token()
-    base_url = get_api_optimisation_url()
+    api_doublon = APIDoublon()
 
-    params = [
-        ('application_ids', identifiant_predemande) for identifiant_predemande in identifiant_predemandes
-    ]
-
-    try:
-        response = requests.get(
-            base_url + 'status',
-            params=params,
-            headers={'x-rdv-opt-auth-token': auth_token},
-            timeout=10,
-        )
-        response.raise_for_status()
-        data = response.json()
-        if not isinstance(data, dict):
-            raise ValueError(f'reponse is not a dict: {data!r}')
-    except (requests.RequestException, ValueError, TypeError) as e:
-        raise ANTSError(repr(e))
+    predemandes_statuses = api_doublon.statuses(identifiant_predemandes)
 
     valid = True
     msg = []
     appointments = []
     for identifiant_predemande in identifiant_predemandes:
-        if identifiant_predemande not in data:
+        if identifiant_predemande not in predemandes_statuses:
             msg.append(f'Prédemande "{identifiant_predemande}" inconnue.')
             valid = False
             continue
-        predemande_state = data[identifiant_predemande]
+        predemande_state = predemandes_statuses[identifiant_predemande]
         if not isinstance(predemande_state, dict):
-            raise ANTSError(f'application_id state is not a dict: {data!r}')
+            raise ANTSError(f'application_id state is not a dict: {predemandes_statuses!r}')
         if predemande_state.get('status') != 'validated':
             msg.append(f'Prédemande "{identifiant_predemande}" inconnue, expirée ou déjà consommée.')
             valid = False
@@ -156,4 +163,4 @@ def get_status_of_predemandes(identifiant_predemandes: list):
             msg.append(f'Prédemande "{identifiant_predemande}" déjà liée à un ou des rendez-vous.')
             valid = False
             continue
-    return valid, ' '.join(msg), data, appointments
+    return valid, ' '.join(msg), predemandes_statuses, appointments