[Sis integration] try to handle failingids

funidata_utils/request_utils/async_httpx_requests.py

@@ -3,6 +3,7 @@
 # ------------------------------------------------------------------------------
 
 import asyncio
+import json
 import logging
 import typing
 from typing import Tuple, Any, Callable, Literal
@@ -73,37 +74,44 @@ async def _binary_search_enabled_post_httpx(
     binary_search_max_depth: int | None = None,
     binary_err_search_sublists: bool = True,
     method: Literal['POST', 'PATCH'] = 'POST',
+    _state: dict[
+                Literal['max_seen_depth', 'sent_requests'], int
+            ] | None = None,
 ) -> list[httpx.Response]:
     is_complex_list_of_batches = False
     if isinstance(payload, list) and all(isinstance(x, list) for x in payload[::3]):
         is_complex_list_of_batches = True
 
     match method:
         case 'POST':
-            _payload = flatten(payload) if is_complex_list_of_batches else payload
-            logger.debug("Sending POST with %d items to %s", len(_payload), path)
+            _request_payload = flatten(payload) if is_complex_list_of_batches else payload
+            logger.debug("Sending POST with %d items to %s", len(_request_payload), path)
             response = await client.post(
                 path,
                 auth=auth,
-                json=_payload,
+                json=_request_payload,
                 params=params,
                 timeout=120,
             )
 
         case 'PATCH':
-            _payload = flatten(payload) if is_complex_list_of_batches else payload
-            logger.debug("Sending PATCH with %d items to %s", len(_payload), path)
+            _request_payload = flatten(payload) if is_complex_list_of_batches else payload
+            logger.debug("Sending PATCH with %d items to %s", len(_request_payload), path)
             response = await client.patch(
                 path,
                 auth=auth,
-                json=_payload,
+                json=_request_payload,
                 params=params,
                 timeout=120,
             )
 
         case _:
             raise Exception(f'Unsupported method: {method}')
 
+    if _state:
+        _state['sent_requests'] = _state.get('sent_requests', 0) + 1
+        _state['max_seen_depth'] = max(_state.get('max_seen_depth', 0), binary_search_depth)
+
     if (
         binary_search_max_depth == 0 or
         (binary_search_max_depth and binary_search_depth >= binary_search_max_depth)
@@ -117,6 +125,7 @@ async def _binary_search_enabled_post_httpx(
     else:
         if len(payload) <= 1 and not binary_err_search_sublists:
             return [response]
+        # Final batch, cannot split into further batches, but can split into sublist if enabled
         if len(payload) == 1 and binary_err_search_sublists:
             return await _binary_search_enabled_post_httpx(
                 path=path,
@@ -128,29 +137,85 @@ async def _binary_search_enabled_post_httpx(
                 binary_search_max_depth=binary_search_max_depth,
                 binary_err_search_sublists=False,
                 method=method,
+                _state=_state
             )
 
+    failing_ids = []
+    if 400 <= response.status_code < 500:
+        # Try to find the "failingIds" from the response of 400-status codes
+        try:
+            err_json = response.json()
+            if isinstance(err_json, str):
+                err_json = json.loads(err_json)
+
+            if err_json.get('failingIds'):
+                failing_ids = err_json['failingIds']
+        except Exception:
+            pass
+
+    # If everything is failed, stop.
+    # _request_payload here is the flattened payload, aka "amount of entities sent equals amount of failing ids"
+    if len(failing_ids) == len(_request_payload):
+        return [response]
+
+    # try to convert the payload into "failed" and "not failed" lists
+    first_batch = []
+    second_batch = []
+    if failing_ids:
+        if is_complex_list_of_batches:
+            # If we are in the context of "complex" aka grouped data: [ [person_1_1, person_1_2], [person_2_1] ]
+            if binary_err_search_sublists:
+                # If we allow searching sublists, we can split entities by passing/failing directly
+                for subset in payload:
+                    for _entity in subset:
+                        if _entity.get('id') in failing_ids:
+                            first_batch.append(_entity)
+                        else:
+                            second_batch.append(_entity)
+            else:
+                # If we don't allow sublist searching, split according to existence of fail in a batch
+                for subset in payload:
+                    subset_ids = {x.get('id') for x in subset}
+                    if any(_id in subset_ids for _id in failing_ids):
+                        first_batch.append(subset)
+                    else:
+                        second_batch.append(subset)
+        else:
+            # Payload is not a list of lists, can directly check against it.
+            for x in payload:
+                if x.get('id') in failing_ids:
+                    first_batch.append(x)
+                else:
+                    second_batch.append(x)
+
+    # If we were unable to create a split at all, continue with default behavior.
+    if len(first_batch) == 0 or len(second_batch) == 0:
+        first_batch = payload[::2]
+        second_batch = payload[1::2]
+
     first_half_responses = await _binary_search_enabled_post_httpx(
         path=path,
-        payload=payload[::2],
+        payload=first_batch,
         auth=auth,
         params=params,
         client=client,
         binary_search_depth=binary_search_depth + 1,
         binary_search_max_depth=binary_search_max_depth,
         binary_err_search_sublists=binary_err_search_sublists,
         method=method,
+        _state=_state,
     )
     second_half_responses = await _binary_search_enabled_post_httpx(
         path=path,
-        payload=payload[1::2],
+        payload=second_batch,
         auth=auth,
         params=params,
         client=client,
         binary_search_depth=binary_search_depth + 1,
         binary_search_max_depth=binary_search_max_depth,
         binary_err_search_sublists=binary_err_search_sublists,
         method=method,
+        _state=_state
     )
     return first_half_responses + second_half_responses
 

tests/helpers.py

@@ -15,6 +15,11 @@ def mock_client():
 def invalid_handler(request: httpx.Request):
     _content = json.loads(request.content)
     _failing_ids = [_x['id'] for _x in _content if _x.get('invalid')]
+    _exception_ids = [_x['id'] for _x in _content if _x.get('exception')]
+    if _exception_ids:
+        return httpx.Response(
+            status_code=500, json={"reason": "HV000029"}
+        )
     if _failing_ids:
         return httpx.Response(
             status_code=422, json={"failingIds": _failing_ids}

tests/test_import_batching_no_sublist.py

@@ -39,15 +39,17 @@ async def test_recursive_import_batching_with_sublists_off_no_fails(mock_client)
         ]
     ]
 
+    _state = {'max_seen_depth': 0}
     results = await _binary_search_enabled_post_httpx(
         path="http://localhost",
         payload=test_data,
-        auth=None,
         client=mock_client,
-        binary_search_depth=0,
         binary_err_search_sublists=False,
         binary_search_max_depth=None,
+        _state=_state,
     )
+    assert _state['max_seen_depth'] == 0
+    assert _state['sent_requests'] == 1
     assert get_entity_counts_by_status_code(results)[200] == 6
     assert get_entity_counts_by_status_code(results).get(422) is None
 
@@ -88,15 +90,17 @@ async def test_recursive_import_batching_with_sublists_off_one_fail(mock_client)
         ]
     ]
 
+    _state = {'max_seen_depth': 0}
     results = await _binary_search_enabled_post_httpx(
         path="http://localhost",
         payload=test_data,
-        auth=None,
         client=mock_client,
-        binary_search_depth=0,
         binary_err_search_sublists=False,
         binary_search_max_depth=None,
+        _state=_state,
     )
+    assert _state['max_seen_depth'] == 1
+    assert _state['sent_requests'] == 3
     assert get_entity_counts_by_status_code(results)[200] == 4
     assert get_entity_counts_by_status_code(results)[422] == 2
 
@@ -139,15 +143,17 @@ async def test_recursive_import_batching_with_sublists_off_multiple_fails(mock_c
         ]
     ]
 
+    _state = {'max_seen_depth': 0}
     results = await _binary_search_enabled_post_httpx(
         path="http://localhost",
         payload=test_data,
-        auth=None,
         client=mock_client,
-        binary_search_depth=0,
         binary_err_search_sublists=False,
         binary_search_max_depth=None,
+        _state=_state,
     )
+    assert _state['max_seen_depth'] == 2
+    assert _state['sent_requests'] == 5
 
     assert get_entity_counts_by_status_code(results)[200] == 2
     assert get_entity_counts_by_status_code(results)[422] == 4
@@ -194,15 +200,144 @@ async def test_recursive_import_batching_with_sublists_off_all_fails(mock_client
         ]
     ]
 
+    _state = {'max_seen_depth': 0}
     results = await _binary_search_enabled_post_httpx(
         path="http://localhost",
         payload=test_data,
-        auth=None,
         client=mock_client,
-        binary_search_depth=0,
         binary_err_search_sublists=False,
         binary_search_max_depth=None,
+        _state=_state,
     )
+    assert _state['max_seen_depth'] == 0
+    assert _state['sent_requests'] == 1
 
     assert get_entity_counts_by_status_code(results).get(200) is None
     assert get_entity_counts_by_status_code(results)[422] == 6
+
+
+@pytest.mark.asyncio
+async def test_recursive_import_batching_with_sublists_off_all_exceptions(mock_client):
+    test_data = [
+        [
+            {
+                "id": 2,
+                "person": 1,
+                "exception": True
+            },
+            {
+                "id": 3,
+                "person": 1,
+                "exception": True,
+            }
+        ],
+        [
+            {
+                "id": 3,
+                "person": 2,
+                "exception": True,
+            },
+            {
+                "id": 4,
+                "person": 2,
+                "exception": True,
+            }
+        ],
+        [
+            {
+                "id": 4,
+                "person": 3,
+                "exception": True,
+            },
+            {
+                "id": 5,
+                "person": 3,
+                "exception": True,
+            }
+        ]
+    ]
+
+    _state = {'max_seen_depth': 0}
+    results = await _binary_search_enabled_post_httpx(
+        path="http://localhost",
+        payload=test_data,
+        client=mock_client,
+        binary_err_search_sublists=False,
+        binary_search_max_depth=None,
+        _state=_state,
+    )
+    assert _state['max_seen_depth'] == 2
+    assert _state['sent_requests'] == 5
+
+
+    assert get_entity_counts_by_status_code(results).get(200) is None
+    assert get_entity_counts_by_status_code(results)[500] == 6
+
+
+@pytest.mark.asyncio
+async def test_recursive_import_batching_with_sublists_off_all_exceptions_limited_max_depth(mock_client):
+    test_data = [
+        [
+            {
+                "id": 2,
+                "person": 1,
+                "exception": True
+            },
+            {
+                "id": 3,
+                "person": 1,
+                "exception": True,
+            }
+        ],
+        [
+            {
+                "id": 3,
+                "person": 2,
+                "exception": True,
+            },
+            {
+                "id": 4,
+                "person": 2,
+                "exception": True,
+            }
+        ],
+        [
+            {
+                "id": 4,
+                "person": 3,
+                "exception": True,
+            },
+            {
+                "id": 5,
+                "person": 3,
+                "exception": True,
+            }
+        ]
+    ]
+
+    _state = {'max_seen_depth': 0}
+    results = await _binary_search_enabled_post_httpx(
+        path="http://localhost",
+        payload=test_data,
+        client=mock_client,
+        binary_err_search_sublists=False,
+        binary_search_max_depth=None,
+        _state=_state,
+    )
+    assert _state['max_seen_depth'] == 2
+
+    _state = {'max_seen_depth': 0}
+    results = await _binary_search_enabled_post_httpx(
+        path="http://localhost",
+        payload=test_data,
+        client=mock_client,
+        binary_err_search_sublists=False,
+        binary_search_max_depth=1,
+        _state=_state,
+    )
+    assert _state['max_seen_depth'] == 1
+    assert _state['sent_requests'] == 3
+
+
+    assert get_entity_counts_by_status_code(results).get(200) is None
+    assert get_entity_counts_by_status_code(results)[500] == 6