Skip to content

Commit

Permalink
Resolves linting issues after merge
Browse files Browse the repository at this point in the history
  • Loading branch information
ChristopherHammond13 committed Oct 16, 2024
1 parent b8646d5 commit 016047e
Showing 1 changed file with 7 additions and 15 deletions.
22 changes: 7 additions & 15 deletions caracara/modules/rtr/batch_session.py
Original file line number Diff line number Diff line change
Expand Up @@ -4,10 +4,10 @@
import logging
from dataclasses import dataclass
from datetime import datetime, timedelta
from itertools import repeat
from functools import partial, wraps
from itertools import repeat
from threading import current_thread
from typing import Dict, List, Tuple, Optional
from typing import Dict, List, Optional, Tuple

from falconpy import RealTimeResponse, RealTimeResponseAdmin

Expand Down Expand Up @@ -45,13 +45,7 @@ class InnerRTRBatchSession: # pylint: disable=too-few-public-methods
a list of InnerRTRBatchSession objects.
"""

def __init__(
self,
batch_id: str,
devices: Dict,
expiry: datetime,
logger: logging.Logger
):
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
self.devices = devices
Expand Down Expand Up @@ -177,7 +171,7 @@ def worker(
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'],
batch_id=response["batch_id"],
devices=successful_devices,
expiry=datetime.now() + timedelta(seconds=SESSION_EXPIRY),
logger=self.logger,
Expand Down Expand Up @@ -257,7 +251,7 @@ def get(
batch_get_cmd_reqs: List[BatchGetCmdRequest] = []
for thread_name, response in completed:
try:
devices = response['combined']['resources']
devices = response["combined"]["resources"]
except KeyError:
self.logger.warning(
"%s | Batch contained no successful get command executions",
Expand Down Expand Up @@ -452,17 +446,15 @@ def run_generic_command( # pylint: disable=too-many-locals
all_responses: Dict = {}
for thread_name, response in completed:
try:
devices = response['combined']['resources']
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)
"%s | Executed commands on a batch of %d hosts", thread_name, len(devices)
)
all_responses.update(devices)

Expand Down

0 comments on commit 016047e

Please sign in to comment.