Normalize code for Ruff and Black checks

config.py

@@ -16,6 +16,7 @@
 
 Please do not modify CLI_VERSION; it is used for debugging purposes.
 """
+
 CLI_VERSION = "v1.0.0"
 
 HOST = ""

core/utils.py

@@ -5,6 +5,7 @@
 
 logger = logging.getLogger("core.engine.utils")
 
+
 def copy_assets(report_path: str) -> None:
     assets_folders = ["css", "img", "icons"]
     assets_path = os.path.join(report_path, "assets")

core/utils_aws.py

@@ -16,7 +16,14 @@
 
 logger = logging.getLogger("core.engine.aws")
 
-def aws_api_call_with_retry(client: Any, function_name: str, parameters: Dict[str, Any], max_retries: int, retry_delay: int) -> Callable[..., Any]:
+
+def aws_api_call_with_retry(
+    client: Any,
+    function_name: str,
+    parameters: Dict[str, Any],
+    max_retries: int,
+    retry_delay: int,
+) -> Callable[..., Any]:
     def api_call(*args, **kwargs):
         for attempt in range(max_retries):
             try:
@@ -26,21 +33,22 @@ def api_call(*args, **kwargs):
                 else:
                     return function_to_call(**kwargs)
             except botocore.exceptions.ClientError as error:
-                error_code = error.response['Error']['Code']
-                #logger.warning(f"ClientError: {error_code}. Attempt {attempt + 1} of {max_retries}. Retrying in {retry_delay} seconds.")
-                if error_code in ['Throttling', 'RequestLimitExceeded']:
-                    time.sleep(retry_delay * (2 ** attempt))
+                error_code = error.response["Error"]["Code"]
+                # logger.warning(f"ClientError: {error_code}. Attempt {attempt + 1} of {max_retries}. Retrying in {retry_delay} seconds.")
+                if error_code in ["Throttling", "RequestLimitExceeded"]:
+                    time.sleep(retry_delay * (2**attempt))
                     continue
                 else:
                     raise
-            except botocore.exceptions.BotoCoreError as error:
-                #logger.warning(f"BotoCoreError: {str(error)}. Attempt {attempt + 1} of {max_retries}. Retrying in {retry_delay} seconds.")
-                time.sleep(retry_delay * (2 ** attempt))
+            except botocore.exceptions.BotoCoreError:
+                # logger.warning(f"BotoCoreError. Attempt {attempt + 1} of {max_retries}. Retrying in {retry_delay} seconds.")
+                time.sleep(retry_delay * (2**attempt))
                 continue
         raise Exception(f"Failed to call {function_name} after {max_retries} attempts")
 
     return api_call  # Return the callable function
 
+
 def convert_datetime(obj: Any) -> Any:
     if isinstance(obj, dict):
         for k, v in obj.items():
@@ -52,7 +60,13 @@ def convert_datetime(obj: Any) -> Any:
         return obj.isoformat()
     return obj
 
-def build_aws_resource_inventory(cloud_service_provider: int, provider_details: Dict[str, Any], report_path: str, raw_data_path: str) -> None:
+
+def build_aws_resource_inventory(
+    cloud_service_provider: int,
+    provider_details: Dict[str, Any],
+    report_path: str,
+    raw_data_path: str,
+) -> None:
     try:
         access_key = provider_details["accessKey"]
         secret_key = provider_details["secretKey"]
@@ -61,7 +75,7 @@ def build_aws_resource_inventory(cloud_service_provider: int, provider_details:
         session = boto3.Session(
             aws_access_key_id=access_key,
             aws_secret_access_key=secret_key,
-            region_name=region
+            region_name=region,
         )
 
         db_path = os.path.join(report_path, "data", "assessment.db")
@@ -80,53 +94,59 @@ def build_aws_resource_inventory(cloud_service_provider: int, provider_details:
         aggregated_resources = defaultdict(int)
 
         # Iterate through each resource type in the JSON
-        for idx, (resource_type_code, resource_info) in enumerate(resource_type_mapping.items(), start=1):
-            parts = resource_type_code.split('.')
+        for idx, (resource_type_code, resource_info) in enumerate(
+            resource_type_mapping.items(), start=1
+        ):
+            parts = resource_type_code.split(".")
             if len(parts) != 4 or parts[0] != "AWS":
-                #logger.warning(f"Invalid resource type format: {resource_type_code}. Skipping.")
+                # logger.warning(f"Invalid resource type format: {resource_type_code}. Skipping.")
                 continue
 
             # Extract service name, operation name, and result key
             service_name, operation_name, result_key = parts[1], parts[2], parts[3]
 
-            #logger.info(f"Processing service {service_name} with operation {operation_name}")
+            # logger.info(f"Processing service {service_name} with operation {operation_name}")
 
             try:
                 client = session.client(service_name, region_name=region)
                 if not hasattr(client, operation_name):
-                    #logger.error(f"Operation {operation_name} does not exist for service {service_name}")
+                    # logger.error(f"Operation {operation_name} does not exist for service {service_name}")
                     continue
 
                 # Make the API call
-                api_call = aws_api_call_with_retry(client, operation_name, {}, max_retries=3, retry_delay=2)
+                api_call = aws_api_call_with_retry(
+                    client, operation_name, {}, max_retries=3, retry_delay=2
+                )
                 response = api_call()
 
                 if isinstance(response, dict):
                     response.pop("ResponseMetadata", None)
                     resources = response.get(result_key.strip(), [])
                     # Handle paginated results
-                    while 'NextToken' in response:
-                        next_token = response['NextToken']
+                    while "NextToken" in response:
+                        next_token = response["NextToken"]
                         response = api_call(NextToken=next_token)
                         response.pop("ResponseMetadata", None)
                         resources.extend(response.get(result_key.strip(), []))
                 else:
-                    #logger.warning(f"No valid response found for {service_name} operation {operation_name}. Skipping.")
+                    # logger.warning(f"No valid response found for {service_name} operation {operation_name}. Skipping.")
                     continue
 
                 # Aggregate the resources
                 for resource in resources:
                     aggregated_resources[(resource_type_code, region)] += 1
 
                 # Store raw data
-                raw_data.append({
-                    "service": service_name,
-                    "operation": operation_name,
-                    "resources": resources
-                })
-
-            except (NoCredentialsError, ClientError, Exception) as e:
-                #logger.error(f"Error while processing {service_name}: {str(e)}", exc_info=True)
+                raw_data.append(
+                    {
+                        "service": service_name,
+                        "operation": operation_name,
+                        "resources": resources,
+                    }
+                )
+
+            except (NoCredentialsError, ClientError, Exception):
+                # logger.error(f"Error while processing {service_name}", exc_info=True)
                 continue
 
         # Save raw data to a JSON file
@@ -140,12 +160,15 @@ def build_aws_resource_inventory(cloud_service_provider: int, provider_details:
         with connect(db_path=db_path) as conn:
             cursor = conn.cursor()
 
-            for (resource_type_code, resource_location), resource_count in aggregated_resources.items():
+            for (
+                resource_type_code,
+                resource_location,
+            ), resource_count in aggregated_resources.items():
                 try:
                     # Map resource type code to resource_type_id
                     resource_info = resource_type_mapping.get(resource_type_code)
                     if not resource_info:
-                        #logger.warning(f"Resource type {resource_type_code} not found in resourcetype mapping. Skipping.")
+                        # logger.warning(f"Resource type {resource_type_code} not found in resourcetype mapping. Skipping.")
                         continue
 
                     resource_type_id = resource_info["id"]
@@ -156,21 +179,31 @@ def build_aws_resource_inventory(cloud_service_provider: int, provider_details:
                         VALUES (?, ?, ?)
                         ON CONFLICT(resource_type, location) DO UPDATE SET count = excluded.count
                         """,
-                        (resource_type_id, resource_location, resource_count)
+                        (resource_type_id, resource_location, resource_count),
                     )
                 except sqlite3.Error as e:
-                    logger.error(f"SQLite error while processing aggregated resource: {e}", exc_info=True)
+                    logger.error(
+                        f"SQLite error while processing aggregated resource: {e}",
+                        exc_info=True,
+                    )
                 except Exception as e:
-                    logger.error(f"Unexpected error while processing aggregated resource: {e}", exc_info=True)
+                    logger.error(
+                        f"Unexpected error while processing aggregated resource: {e}",
+                        exc_info=True,
+                    )
 
             conn.commit()
 
     except Exception as e:
         logger.error(f"Error creating AWS resource inventory: {str(e)}", exc_info=True)
 
+
 def get_missing_months_aws(processed_costs: Set[str], max_months: int) -> List[date]:
     current_date = datetime.utcnow().date().replace(day=1)
-    processed_months = {datetime.strptime(month_str, '%Y-%m-%d').date().replace(day=1) for month_str in processed_costs}
+    processed_months = {
+        datetime.strptime(month_str, "%Y-%m-%d").date().replace(day=1)
+        for month_str in processed_costs
+    }
     missing_months = []
 
     for i in range(max_months):
@@ -180,46 +213,66 @@ def get_missing_months_aws(processed_costs: Set[str], max_months: int) -> List[d
 
     return missing_months
 
-def build_aws_cost_inventory(cloud_service_provider: int, provider_details: Dict[str, Any], report_path: str, raw_data_path: str) -> None:
+
+def build_aws_cost_inventory(
+    cloud_service_provider: int,
+    provider_details: Dict[str, Any],
+    report_path: str,
+    raw_data_path: str,
+) -> None:
     try:
         session = boto3.Session(
             aws_access_key_id=provider_details["accessKey"],
             aws_secret_access_key=provider_details["secretKey"],
-            region_name=provider_details["region"]
+            region_name=provider_details["region"],
         )
-        cost_explorer = session.client('ce', region_name='us-east-1')
+        cost_explorer = session.client("ce", region_name="us-east-1")
 
         db_path = os.path.join(report_path, "data", "assessment.db")
 
         end_time = date.today().replace(day=1) + relativedelta(months=1)
         start_time = end_time - relativedelta(months=6)
 
         cost_and_usage = cost_explorer.get_cost_and_usage(
-            TimePeriod={'Start': start_time.strftime('%Y-%m-%d'), 'End': end_time.strftime('%Y-%m-%d')},
-            Granularity='MONTHLY',
-            Metrics=['UnblendedCost'],
-            GroupBy=[{'Type': 'DIMENSION', 'Key': 'SERVICE'}],
+            TimePeriod={
+                "Start": start_time.strftime("%Y-%m-%d"),
+                "End": end_time.strftime("%Y-%m-%d"),
+            },
+            Granularity="MONTHLY",
+            Metrics=["UnblendedCost"],
+            GroupBy=[{"Type": "DIMENSION", "Key": "SERVICE"}],
             Filter={
-                'Dimensions': {
-                    'Key': 'REGION',
-                    'Values': [provider_details["region"]]
-                }
-            }
+                "Dimensions": {"Key": "REGION", "Values": [provider_details["region"]]}
+            },
         )
 
-        cost_inventory_raw_path = os.path.join(raw_data_path, "cost_inventory_raw_data.json")
+        cost_inventory_raw_path = os.path.join(
+            raw_data_path, "cost_inventory_raw_data.json"
+        )
         with open(cost_inventory_raw_path, "w", encoding="utf-8") as raw_file:
             json.dump(cost_and_usage, raw_file, indent=4)
 
         # Insert structured data into SQLite
         with connect(db_path=db_path) as conn:
             cursor = conn.cursor()
 
-            for result in cost_and_usage['ResultsByTime']:
-                month_str = result['TimePeriod']['Start']
-                total_cost = sum(float(group['Metrics']['UnblendedCost']['Amount']) for group in result['Groups'])
-                currency = result['Groups'][0]['Metrics']['UnblendedCost']['Unit'] if result['Groups'] else 'USD'
-                month_date = datetime.strptime(month_str, '%Y-%m-%d').date().replace(day=1).isoformat()
+            for result in cost_and_usage["ResultsByTime"]:
+                month_str = result["TimePeriod"]["Start"]
+                total_cost = sum(
+                    float(group["Metrics"]["UnblendedCost"]["Amount"])
+                    for group in result["Groups"]
+                )
+                currency = (
+                    result["Groups"][0]["Metrics"]["UnblendedCost"]["Unit"]
+                    if result["Groups"]
+                    else "USD"
+                )
+                month_date = (
+                    datetime.strptime(month_str, "%Y-%m-%d")
+                    .date()
+                    .replace(day=1)
+                    .isoformat()
+                )
 
                 # Insert or update the cost data for the month
                 cursor.execute(
@@ -230,12 +283,17 @@ def build_aws_cost_inventory(cloud_service_provider: int, provider_details: Dict
                         cost = excluded.cost,
                         currency = excluded.currency
                     """,
-                    (month_date, total_cost, currency)
+                    (month_date, total_cost, currency),
                 )
 
             # Handle missing months
-            structured_months = {datetime.strptime(result['TimePeriod']['Start'], '%Y-%m-%d').date() for result in cost_and_usage['ResultsByTime']}
-            missing_months = get_missing_months_aws({month.isoformat() for month in structured_months}, 6)
+            structured_months = {
+                datetime.strptime(result["TimePeriod"]["Start"], "%Y-%m-%d").date()
+                for result in cost_and_usage["ResultsByTime"]
+            }
+            missing_months = get_missing_months_aws(
+                {month.isoformat() for month in structured_months}, 6
+            )
 
             for missing_month in missing_months:
                 cursor.execute(
@@ -245,7 +303,7 @@ def build_aws_cost_inventory(cloud_service_provider: int, provider_details: Dict
                     ON CONFLICT(month) DO UPDATE SET
                         currency = excluded.currency
                     """,
-                    (missing_month.isoformat(), currency)
+                    (missing_month.isoformat(), currency),
                 )
 
             conn.commit()