Code changes

dataplex-quickstart-labs/00-resources/scripts/python/business-glossary-import/bg_import/api_call_utils.py

@@ -132,29 +132,42 @@ def fetch_api_response(
   """
   data, error_msg = None, None
   method_name = 'GET' if method == requests.get else 'POST'
+  context = f'[{method_name} {url}]'
+
+  logger.debug(
+    f'{context} Initiating call with project_id: {project_id} and request_body: {request_body}'
+  )
+
   try:
     res = method(url, headers=_get_header(project_id), json=request_body)
+    logger.debug(f'{context} Response status: {res.status_code}, Response text: {res.text}')
+
     try:
       data = res.json()
     except requests.exceptions.JSONDecodeError:
-      error_msg = f'{method_name} call to {url} returned non valid JSON.'
+      error_msg = 'Call returned non-valid JSON.'
+      logger.debug(f'{context} Json decode error: {error_msg}')
       return {
           'json': None,
           'error_msg': error_msg
       }
+
     if not res.ok:
       # If the response is an error, capture the error message from the JSON
-      error_msg = data.get('error', {}).get('message') or f'{method_name} call to {url} returned HTTP {res.status_code}.'
+      error_msg = data.get('error', {}).get('message') or f'Call returned HTTP {res.status_code}.'
+      logger.debug(f'{context} Bad response error: {error_msg}')
       return {
           'json': data,
           'error_msg': error_msg
       }
+    logger.debug(f'Successful response JSON: {data}')
     return {
         'json': data,
         'error_msg': None
     }
   except requests.exceptions.RequestException as err:
     error_msg = create_error_message(method_name, url, err)
+    logger.debug(f'{context} Exception occurred: {error_msg}')
     return {
         'json': None,
         'error_msg': error_msg

dataplex-quickstart-labs/00-resources/scripts/python/business-glossary-import/bg_import/business_glossary_export_v2.py

@@ -161,36 +161,35 @@ def compute_ancestors(
     map_entry_id_to_entry_type: Dict[str, str],
 ) -> List[Dict[str, str]]:
     """
-    Build the ancestors array for an entry (using entry IDs).
-    For each ancestor in the chain, construct its export resource using get_export_resource_by_id.
-    If no belongs_to exists, assume the glossary is the only ancestor.
+    Builds an ancestors array containing only the root glossary and the immediate parent.
 
     Args:
         child_id (str): The entry id of the child.
         parent_mapping (Dict[str, str]): The parent mapping.
         map_entry_id_to_entry_type (Dict[str, str]): The mapping of entry id to entry type.
     Returns:
-        List[Dict[str, str]]: The ancestors array.
+        List[Dict[str, str]]: The ancestors array with at most two elements.
     """
     ancestors: List[Dict[str, str]] = []
-    if child_id in parent_mapping:
-        current = parent_mapping[child_id]
-        while True:
-            current_type = map_entry_id_to_entry_type.get(current, "glossary")
-            resource = get_export_resource_by_id(current, current_type)
-            glossary_child_entry_name = f"{DATAPLEX_ENTRY_GROUP}/entries/{resource}"
-            ancestors.append({"name": glossary_child_entry_name, "type": get_entry_type_name(current_type)})
-            if current not in parent_mapping:
-                break
-            current = parent_mapping[current]
-
     glossary_entry_name = (
         f"{DATAPLEX_ENTRY_GROUP}/entries/projects/{PROJECT}/locations/{GLOSSARY_EXPORT_LOCATION}/glossaries/{GLOSSARY}"
     )
-    ancestors.append({"name": glossary_entry_name, "type": get_entry_type_name("glossary")})
-    ancestors.reverse()
-    return ancestors
+    ancestors.append({
+        "name": glossary_entry_name, 
+        "type": get_entry_type_name("glossary")
+    })
+
+    if child_id in parent_mapping:
+        parent_id = parent_mapping[child_id]
+        parent_type = map_entry_id_to_entry_type.get(parent_id, "glossary_category")    
+        resource = get_export_resource_by_id(parent_id, parent_type)
+        parent_entry_name = f"{DATAPLEX_ENTRY_GROUP}/entries/{resource}"   
+        ancestors.append({
+            "name": parent_entry_name,
+            "type": get_entry_type_name(parent_type)
+        })
 
+    return ancestors
 
 def process_entry(
     entry: Dict[str, Any],
@@ -486,6 +485,9 @@ def process_term_entry_links(entry: Dict[str, Any]) -> List[Dict[str, Any]]:
             rel_url = f"https://datacatalog.googleapis.com/v2/{relative_resource_name}/relationships"
             response = api_call_utils.fetch_api_response(requests.get, rel_url, USER_PROJECT)
             logger.debug(f"Relationships response for {rel_url}: {response}")
+            if not response or not response.get("json"):
+                logger.warning(f"No relationships found for {relative_resource_name}")
+                continue
             relationships = response.get("json", {}).get("relationships", [])
 
             for rel in relationships: