Hierarchial crawl for in place archives

estimators/eo_in_place_archive_estimator.py

@@ -214,7 +214,7 @@ def parse_and_count_in_place_archive_mail_box(self, mail_box_ids: List[str], fai
 
             for batch_id, responses in next_response_map.items():
                 batch = next_batch_id_to_batch_map[batch_id]
-                new_pending_next_items.extend(process_pagination_responses(batch, responses, mailbox_to_resp_map, "mailboxId", GRAPH_BETA_URL, failures, False))
+                new_pending_next_items.extend(process_pagination_responses(batch, responses, mailbox_to_resp_map, "mailboxId", GRAPH_BETA_URL, failures, True))
 
             pending_next_items = new_pending_next_items
 

estimators/factory.py

@@ -15,6 +15,7 @@ def __init__(
     self,
     config: ScanConfig,
     manager: TokenManager = None,
+    one_token_per_app_manager = None,
     logger = None,
     stop_event = None,
     id_to_display_name = None
@@ -31,6 +32,8 @@ def __init__(
     self.url_invoker = None
     self.chat_estimator = None
     self.mock_url_invoker = None
+    self.one_token_per_app_manager = one_token_per_app_manager
+    self.child_folder_url_invoker = None
 
   def isEmpty(self, data):
     return data is None or len(data) == 0
@@ -62,6 +65,16 @@ def get_url_invoker(self, hard_reset=False):
       )
 
     return self.url_invoker
+
+  def get_child_folder_url_invoker(self, hard_reset=False):
+    if self.child_folder_url_invoker is None or hard_reset:
+      if self.one_token_per_app_manager is None:
+        raise Exception("One token per app manager not initialized!!")
+      self.child_folder_url_invoker = UrlInvoker(
+        self.one_token_per_app_manager, self.config.retries, self.config.backoff, 1, 0.5
+      )
+
+    return self.child_folder_url_invoker
 
   def get_mock_url_invoker(self, hard_reset=False, seed=None):
     if self.mock_url_invoker is None or hard_reset:
@@ -120,10 +133,11 @@ def get_in_place_archive_estimator(
       if self.in_place_archive_estimator is not None:
         self.in_place_archive_estimator.shutdown()
       url_invoker = self.get_url_invoker()
+      child_folder_url_invoker = self.get_child_folder_url_invoker()
       self.in_place_archive_estimator = EOInPlaceArchiveEstimator(
           self.config,
           url_invoker,
-          None,  # TODO Add support for non delta API based flow through factory
+          child_folder_url_invoker,
           logger=self.logger,
           stop_event=self.stop_event,
           use_delta_api=use_delta_api,

ui/chats_ui.py

@@ -1606,7 +1606,7 @@ def execute_migration_scan(self):
       manager = self._authenticate_if_needed(config)
 
       self.factory = EstimatorFactory(
-          config, manager, self.log_msg, self.stop_scan_event, None
+          config, manager, None, self.log_msg, self.stop_scan_event, None
       )
 
       # 3. User Discovery

ui/exchange_online_ui.py

@@ -1403,7 +1403,7 @@ def execute_migration_scan(self):
       one_token_per_app_manager = None
       if config.scan_in_place_archives:
         one_token_per_app_manager = self._authenticate_if_needed(config, use_single_app=True)
-      self.factory = EstimatorFactory(config, manager, self.log_msg, self.stop_scan_event, None)
+      self.factory = EstimatorFactory(config, manager, one_token_per_app_manager, self.log_msg, self.stop_scan_event, None)
 
       # 3. User Discovery
       all_users, existing_data, groups, shared_mailboxes = self._resolve_target_users(config, manager, one_token_per_app_manager)
@@ -1802,7 +1802,7 @@ def run_batch_scan(
     executor = ThreadPoolExecutor(max_workers=config.concurrency)
     estimator: Estimator = None
     if resource_type == "in_place_archives":
-      estimator = self.factory.get_in_place_archive_estimator(use_delta_api=True)
+      estimator = self.factory.get_in_place_archive_estimator(use_delta_api=False)
     elif resource_type == "shared_mails":
       estimator = self.factory.get_shared_mailbox_estimator()
 

util/utils.py

@@ -191,10 +191,10 @@ def process_pagination_responses(
 
     for req in batch:
         req_id = req["id"]
+        key = req["headers"][grouping_key]
+
         if req_id in batch_responses_map:
-            resp = batch_responses_map[req_id]
-            key = req["headers"][grouping_key]
-
+            resp = batch_responses_map[req_id] 
             # Retrieve original response object
             orig_entry = orig_map[key]
             orig_resp = orig_entry["resp"] if isinstance(orig_entry, dict) and "resp" in orig_entry else orig_entry
@@ -220,6 +220,14 @@ def process_pagination_responses(
                     "statusCode" : resp["status"],
                     "message": resp["body"]["error"]["message"]
                 })
+        else:
+          failures.append({
+            "mailboxId": key,
+            "isPartial": is_partial,
+            "type": FailureType.NOT_FOUND,
+            "statusCode": None,
+            "message": "No response found for API Call."
+          })
 
     return next_items