RTR Robustness Updates (#205)

* Resolves #187 and some other general RTR snags. Also bumps dependencies.

* Resolves linting issues

* Caracara 0.9.0: Preparation for Python 3.13 (and huge code cleanup) (#206)

* Major code formatting cleanup, deprecation of Python 3.8, and removal of setuptools requirement

* Add note about Python version compatibility

* Temporarily set Python 3.13 CI version to 3.13 RC2

* Changes to sate isort and black together

* Unified development dependencies

* Bumps py7zr to 0.22.0 as we no longer support Python 3.7. Also, bumps caracara-filters to 1.0.0+.

* Bumps dependencies

* Resolves #187 and some other general RTR snags. Also bumps dependencies.

* Resolves linting issues

* Resolves linting issues after merge

caracara/modules/rtr/batch_session.py

@@ -5,8 +5,9 @@
 from dataclasses import dataclass
 from datetime import datetime, timedelta
 from functools import partial, wraps
+from itertools import repeat
 from threading import current_thread
-from typing import Dict, List
+from typing import Dict, List, Optional, Tuple
 
 from falconpy import RealTimeResponse, RealTimeResponseAdmin
 
@@ -44,11 +45,6 @@ class InnerRTRBatchSession:  # pylint: disable=too-few-public-methods
     a list of InnerRTRBatchSession objects.
     """
 
-    batch_id: str = None
-    devices: Dict = None
-    expiry: datetime = None
-    logger: logging.Logger = None
-
     def __init__(self, batch_id: str, devices: Dict, expiry: datetime, logger: logging.Logger):
         """Configure an inner batch of RTR sessions."""
         self.batch_id = batch_id
@@ -74,7 +70,7 @@ def generic_rtr_worker(
     func: partial,
     logger: logging.Logger,
     device_ids: List[str] = None,
-):
+) -> Tuple[str, Dict]:
     """
     Execute a partial function against an RTR batch session.
 
@@ -96,7 +92,7 @@ def generic_rtr_worker(
         func.keywords["optional_hosts"] = device_ids_in_batch
     response = func(batch_id=session.batch_id)["body"]
     logger.debug("%s | %s", thread_name, response)
-    return response
+    return thread_name, response
 
 
 class RTRBatchSession:
@@ -121,14 +117,14 @@ class RTRBatchSession:
     @_batch_session_required
     def device_ids(self) -> List[str]:
         """Return a list of device IDs from all inner batch sessions."""
-        return [x.devices.keys() for x in self.batch_sessions]
+        return [x.devices for x in self.batch_sessions]
 
     def connect(
         self,
         device_ids: List[str],
         queueing: bool = False,
         timeout: int = default_timeout,
-    ):
+    ) -> bool:
         """
         Establish a connection to one or more hosts.
 
@@ -142,24 +138,45 @@ def connect(
             batches.append(device_ids[i : i + MAX_BATCH_SESSION_HOSTS])
         self.logger.info("Divided up devices into %d batches", len(batches))
 
-        def worker(batch_device_ids: List[str], batch_func: partial):
+        def worker(
+            batch_device_ids: List[str],
+            batch_func: partial,
+        ) -> Tuple[str, Optional[InnerRTRBatchSession]]:
             thread_name = current_thread().name
             self.logger.info(
                 "%s | Batch worker started with a list of %d devices",
                 thread_name,
                 len(batch_device_ids),
             )
             response = batch_func(host_ids=batch_device_ids)["body"]
+            self.logger.debug("%s | %s", thread_name, str(response))
             resources = response["resources"]
-            self.logger.info("%s | Connected to %s systems", thread_name, len(resources))
+
+            # Identify devices that failed to connect and/or returned an error
+            # Resolves GitHub issue #187
+            if not resources:
+                self.logger.info("%s | Resource list is empty", thread_name)
+                return thread_name, None
+
+            successful_devices = {
+                device_id: device_data
+                for device_id, device_data in resources.items()
+                if not device_data.get("errors")
+            }
+
+            if not successful_devices:
+                self.logger.info("%s | Successful device list is empty", thread_name)
+                return thread_name, None
+
+            self.logger.info("%s | Connected to %s systems", thread_name, len(successful_devices))
             self.logger.debug("%s | %s", thread_name, response)
             batch_data = InnerRTRBatchSession(
                 batch_id=response["batch_id"],
-                devices=resources,
+                devices=successful_devices,
                 expiry=datetime.now() + timedelta(seconds=SESSION_EXPIRY),
                 logger=self.logger,
             )
-            return batch_data
+            return thread_name, batch_data
 
         batch_func = partial(
             self.api.batch_init_sessions,
@@ -171,17 +188,27 @@ def worker(batch_device_ids: List[str], batch_func: partial):
         with concurrent.futures.ThreadPoolExecutor(
             max_workers=MAX_BATCH_SESSION_THREADS
         ) as executor:
-            completed = executor.map(worker, batches, [batch_func])
+            completed = executor.map(worker, batches, repeat(batch_func))
 
         self.batch_sessions = []
-        for complete in completed:
-            self.logger.info("Completed a batch of RTR connections")
-            self.batch_sessions.append(complete)
+        for thread_name, thread_data in completed:
+            self.logger.info("%s | Completed a batch of RTR connections", thread_name)
+            if thread_data is None:
+                self.logger.info("%s | Batch contained no successful connections", thread_name)
+            else:
+                self.logger.info(
+                    "%s | Batch contained %d successful connections",
+                    thread_name,
+                    len(thread_data.devices),
+                )
+                self.batch_sessions.append(thread_data)
 
         device_count = sum(len(d.devices) for d in self.batch_sessions)
         self.logger.info("Connected to %d devices", device_count)
         self.logger.debug(self.batch_sessions)
 
+        return len(self.batch_sessions) > 0
+
     @_batch_session_required
     def disconnect(self):
         """Disconnect the RTR batch session."""
@@ -218,15 +245,29 @@ def get(
             completed: List[Dict] = executor.map(
                 partial_worker,
                 self.batch_sessions,
-                [partial_get_func],
+                repeat(partial_get_func),
             )
 
         batch_get_cmd_reqs: List[BatchGetCmdRequest] = []
-        for complete in completed:
-            self.logger.info("Executed commands on a batch of %d hosts", len(complete))
+        for thread_name, response in completed:
+            try:
+                devices = response["combined"]["resources"]
+            except KeyError:
+                self.logger.warning(
+                    "%s | Batch contained no successful get command executions",
+                    thread_name,
+                )
+                continue
+
+            self.logger.info(
+                "%s | Executed get command on a batch of %d devices",
+                thread_name,
+                len(devices),
+            )
+
             batch_get_cmd_req = BatchGetCmdRequest(
-                batch_get_cmd_req_id=complete["batch_get_cmd_req_id"],
-                devices=complete["combined"]["resources"],
+                batch_get_cmd_req_id=response["batch_get_cmd_req_id"],
+                devices=devices,
             )
             batch_get_cmd_reqs.append(batch_get_cmd_req)
 
@@ -278,7 +319,7 @@ def worker(batch_get_cmd_req: BatchGetCmdRequest, func: partial) -> List[GetFile
             timeout_duration=f"{timeout}s",
         )
         with concurrent.futures.ThreadPoolExecutor(max_workers=threads) as executor:
-            completed = executor.map(worker, batch_get_cmd_reqs, [partial_func])
+            completed = executor.map(worker, batch_get_cmd_reqs, repeat(partial_func))
 
         all_get_files: List[GetFile] = []
 
@@ -345,13 +386,13 @@ def worker(session: InnerRTRBatchSession, func: partial):
         with concurrent.futures.ThreadPoolExecutor(
             max_workers=MAX_BATCH_SESSION_THREADS
         ) as executor:
-            completed = executor.map(worker, self.batch_sessions, [batch_func])
+            completed = executor.map(worker, self.batch_sessions, repeat(batch_func))
 
         for complete in completed:
             self.logger.info("Refreshed session %s", complete)
 
     @_batch_session_required
-    def run_generic_command(
+    def run_generic_command(  # pylint: disable=too-many-locals
         self,
         command_string: str,
         device_ids: List[str] = None,
@@ -399,13 +440,23 @@ def run_generic_command(
             completed: List[Dict] = executor.map(
                 partial_worker,
                 self.batch_sessions,
-                [partial_cmd_func],
+                repeat(partial_cmd_func),
             )
 
         all_responses: Dict = {}
-        for complete in completed:
-            self.logger.info("Executed commands on a batch of %d hosts", len(complete))
-            all_responses.update(complete["combined"]["resources"])
+        for thread_name, response in completed:
+            try:
+                devices = response["combined"]["resources"]
+            except KeyError:
+                self.logger.warning(
+                    "%s | Batch contained no successful command executions",
+                    thread_name,
+                )
+                continue
+            self.logger.info(
+                "%s | Executed commands on a batch of %d hosts", thread_name, len(devices)
+            )
+            all_responses.update(devices)
 
         return all_responses