Merge pull request #36 from bento-platform/fix-handling-bad-table-res…

…ponse

Fix handling of bad table responses

.github/workflows/build.yml

@@ -22,7 +22,7 @@ jobs:
         uses: actions/checkout@v3
 
       - name: Run Bento build action
-        uses: bento-platform/bento_build_action@v0.9.3
+        uses: bento-platform/bento_build_action@v0.10.1
         with:
           registry: ghcr.io
           registry-username: ${{ github.actor }}

bento_aggregation_service/search/dataset_search.py

@@ -172,27 +172,35 @@ async def _fetch_table_definition_worker(table_queue: Queue, auth_header: Option
             # Exit signal
             return
 
+        service_kind = t["service_artifact"]  # TODO: this should eventually be migrated to service_kind or something...
+
         try:
             # - Gohan compatibility
             # TODO: formalize/clean this up
-            artifact = t["service_artifact"]
-            if USE_GOHAN and artifact == "variant":
-                artifact = "gohan"
+            if USE_GOHAN and service_kind == "variant":
+                service_kind = "gohan"
 
             # Setup up pre-requisites
-            url = f"api/{artifact}/tables/{t['table_id']}"
+            url = f"api/{service_kind}/tables/{t['table_id']}"
 
             logger.debug(f"Table URL fragment: {url}")
 
             # TODO: Don't fetch schema except for first time?
-            table_ownerships_and_records.append((t, await bento_fetch(
+            table_record = await bento_fetch(
                 client,
                 url,
                 method="GET",
                 auth_header=auth_header,  # Required, otherwise may hit a 403 error
                 extra_headers=DATASET_SEARCH_HEADERS
-            )))
-            # TODO: Handle HTTP errors
+            )
+
+            if isinstance(table_record, dict):
+                table_ownerships_and_records.append((t, table_record))
+            else:
+                logger.error(f"Encountered malformatted table record from {service_kind} (will skip): {table_record}")
+
+        except Exception as e:
+            logger.error(f"Encountered error while trying to fetch table record from {service_kind}: {e}")
 
         finally:
             table_queue.task_done()

bento_aggregation_service/utils.py

@@ -23,9 +23,15 @@
 RequestBody = Optional[Union[bytes, str]]
 
 
-async def bento_fetch(client: AsyncHTTPClient, path_fragment: str, request_body: RequestBody = None,
-                      method: str = "POST", auth_header: Optional[str] = None, extra_headers: Optional[dict] = None,
-                      url_args: Tuple[Tuple[str, str]] = ()):
+async def bento_fetch(
+    client: AsyncHTTPClient,
+    path_fragment: str,
+    request_body: RequestBody = None,
+    method: str = "POST",
+    auth_header: Optional[str] = None,
+    extra_headers: Optional[dict] = None,
+    url_args: Tuple[Tuple[str, str]] = (),
+) -> Union[None, list, dict, int, float, str, bool]:
     if isinstance(request_body, str):
         # Convert str to bytes with only accepted charset: UTF-8
         request_body = request_body.encode("UTF-8")