diff --git a/core/engine.py b/core/engine.py index 441e099..ca59b9f 100644 --- a/core/engine.py +++ b/core/engine.py @@ -2,8 +2,8 @@ import logging import os import boto3 -from datetime import datetime -from typing import Any, Dict, Optional, Tuple +from datetime import datetime, timezone +from typing import Any from azure.identity import ClientSecretCredential from azure.mgmt.resource import ResourceManagementClient from azure.core.exceptions import ClientAuthenticationError @@ -27,8 +27,8 @@ # Stage 1 def verify_credentials( - cloud_service_provider: int, provider_details: Dict[str, Any] -) -> Tuple[bool, str]: + cloud_service_provider: int, provider_details: dict[str, Any] +) -> tuple[bool, str]: connection_success = False logs = "" @@ -78,8 +78,8 @@ def verify_credentials( # Stage 2 def test_permissions( - cloud_service_provider: int, provider_details: Dict[str, Any] -) -> Tuple[bool, bool, bool, str]: + cloud_service_provider: int, provider_details: dict[str, Any] +) -> tuple[bool, bool, bool, str]: permission_valid = False permission_reader = False permission_cost = False @@ -182,10 +182,10 @@ def test_permissions( # Stage 3 def create_resource_inventory( cloud_service_provider: int, - provider_details: Dict[str, Any], + provider_details: dict[str, Any], report_path: str, raw_data_path: str, -) -> Dict[str, Any]: +) -> dict[str, Any]: # Copy assets and datasets folders data copy_assets(report_path) @@ -211,10 +211,10 @@ def create_resource_inventory( # Stage 4 def create_cost_inventory( cloud_service_provider: int, - provider_details: Dict[str, Any], + provider_details: dict[str, Any], report_path: str, raw_data_path: str, -) -> Dict[str, Any]: +) -> dict[str, Any]: try: if cloud_service_provider == 1: # Azure build_azure_cost_inventory( @@ -237,10 +237,10 @@ def sync_assessment( report_path: str, name: str, started_at: int, - metadata: Dict[str, Any], + metadata: dict[str, Any], mode: str, - token: Optional[str], -) -> Dict[str, Any]: + token: str | None, +) -> dict[str, Any]: if mode != "online" or not token: return { "success": True, @@ -323,7 +323,7 @@ def sync_assessment( # Stage 5 - Offline def perform_risk_assessment( exit_strategy: int, report_path: str, mode: str -) -> Dict[str, Any]: +) -> dict[str, Any]: if mode != "offline": logger.debug("Online mode – skipping local risk assessment.") @@ -420,13 +420,13 @@ def perform_risk_assessment( # Stage 6 def generate_report( cloud_service_provider: int, - provider_details: Dict[str, Any], + provider_details: dict[str, Any], exit_strategy: int, assessment_type: int, name: str, report_path: str, raw_data_path: str, -) -> Dict[str, Any]: +) -> dict[str, Any]: try: db_path = os.path.join(report_path, "data", "assessment.db") @@ -443,7 +443,7 @@ def generate_report( scoring_data = load_data("scoring_data", db_path=db_path) # Timestamp - timestamp = datetime.utcnow().strftime("%Y-%m-%d %H:%M:%S UTC") + timestamp = datetime.now(timezone.utc).strftime("%Y-%m-%d %H:%M:%S UTC") metadata = { "name": name, diff --git a/core/utils_aws.py b/core/utils_aws.py index ab31aba..c951aaa 100644 --- a/core/utils_aws.py +++ b/core/utils_aws.py @@ -6,8 +6,9 @@ import time import logging import sqlite3 -from typing import Any, Dict, Set, List, Callable -from datetime import date, datetime +from collections.abc import Callable +from typing import Any +from datetime import date, datetime, timezone from collections import defaultdict from dateutil.relativedelta import relativedelta from botocore.exceptions import NoCredentialsError, ClientError @@ -20,7 +21,7 @@ def aws_api_call_with_retry( client: Any, function_name: str, - parameters: Dict[str, Any], + parameters: dict[str, Any], max_retries: int, retry_delay: int, ) -> Callable[..., Any]: @@ -63,7 +64,7 @@ def convert_datetime(obj: Any) -> Any: def build_aws_resource_inventory( cloud_service_provider: int, - provider_details: Dict[str, Any], + provider_details: dict[str, Any], report_path: str, raw_data_path: str, ) -> None: @@ -198,8 +199,8 @@ def build_aws_resource_inventory( 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) +def get_missing_months_aws(processed_costs: set[str], max_months: int) -> list[date]: + current_date = datetime.now(timezone.utc).date().replace(day=1) processed_months = { datetime.strptime(month_str, "%Y-%m-%d").date().replace(day=1) for month_str in processed_costs @@ -216,7 +217,7 @@ def get_missing_months_aws(processed_costs: Set[str], max_months: int) -> List[d def build_aws_cost_inventory( cloud_service_provider: int, - provider_details: Dict[str, Any], + provider_details: dict[str, Any], report_path: str, raw_data_path: str, ) -> None: diff --git a/core/utils_azure.py b/core/utils_azure.py index b69b534..53632f1 100644 --- a/core/utils_azure.py +++ b/core/utils_azure.py @@ -3,7 +3,7 @@ import os import logging import sqlite3 -from typing import Any, Dict, Set +from typing import Any from datetime import date, datetime from dateutil.relativedelta import relativedelta from collections import defaultdict @@ -43,7 +43,7 @@ def is_resource_inventory_empty( def build_azure_resource_inventory( cloud_service_provider: int, - provider_details: Dict[str, Any], + provider_details: dict[str, Any], report_path: str, raw_data_path: str, ) -> None: @@ -133,7 +133,7 @@ def build_azure_resource_inventory( logger.error(f"Error fetching Azure resources: {str(e)}", exc_info=True) -def get_missing_months_azure(processed_costs: Set[str], months_back: int) -> Set[date]: +def get_missing_months_azure(processed_costs: set[str], months_back: int) -> set[date]: today = date.today() start_date = today.replace(day=1) - relativedelta(months=months_back - 1) all_months = { @@ -158,7 +158,7 @@ def get_missing_months_azure(processed_costs: Set[str], months_back: int) -> Set def build_azure_cost_inventory( cloud_service_provider: int, - provider_details: Dict[str, Any], + provider_details: dict[str, Any], report_path: str, raw_data_path: str, ) -> None: diff --git a/core/utils_report.py b/core/utils_report.py index 8b8d306..77dc9d8 100644 --- a/core/utils_report.py +++ b/core/utils_report.py @@ -2,7 +2,7 @@ import os import json import logging -from typing import List, Dict, Any, Optional +from typing import Any from jinja2 import Template # ReportLab @@ -63,15 +63,15 @@ def anonymize_string(s: str, num_visible: int = 4) -> str: def generate_html_report( report_path: str, - metadata: Dict[str, Any], - resource_type_mapping: Dict[str, Dict[str, Any]], - resource_inventory: List[Dict[str, Any]], - cost_data: List[Dict[str, Any]], - scoring_data: Optional[Dict[str, Any]], - risk_data: List[Dict[str, Any]], - risk_definitions: List[Dict[str, Any]], - alternatives: List[Dict[str, Any]], - alternative_technologies: List[Dict[str, Any]], + metadata: dict[str, Any], + resource_type_mapping: dict[str, dict[str, Any]], + resource_inventory: list[dict[str, Any]], + cost_data: list[dict[str, Any]], + scoring_data: dict[str, Any] | None, + risk_data: list[dict[str, Any]], + risk_definitions: list[dict[str, Any]], + alternatives: list[dict[str, Any]], + alternative_technologies: list[dict[str, Any]], exit_strategy: int, ) -> str: @@ -161,15 +161,15 @@ def generate_html_report( def generate_json_report( raw_data_path: str, - metadata: Dict[str, Any], - resource_type_mapping: Dict[str, Dict[str, Any]], - resource_inventory: List[Dict[str, Any]], - cost_data: List[Dict[str, Any]], - scoring_data: Optional[Dict[str, Any]], - risk_data: List[Dict[str, Any]], - risk_definitions: List[Dict[str, Any]], - alternatives: List[Dict[str, Any]], - alternative_technologies: List[Dict[str, Any]], + metadata: dict[str, Any], + resource_type_mapping: dict[str, dict[str, Any]], + resource_inventory: list[dict[str, Any]], + cost_data: list[dict[str, Any]], + scoring_data: dict[str, Any] | None, + risk_data: list[dict[str, Any]], + risk_definitions: list[dict[str, Any]], + alternatives: list[dict[str, Any]], + alternative_technologies: list[dict[str, Any]], exit_strategy: int, ) -> str: # Transform data for JSON @@ -215,17 +215,17 @@ def generate_json_report( def generate_pdf_report( - provider_details: Dict[str, Any], + provider_details: dict[str, Any], report_path: str, - metadata: Dict[str, Any], - resource_type_mapping: Dict[str, Any], - resource_inventory: List[Dict[str, Any]], - cost_data: List[Dict[str, Any]], - scoring_data: Optional[Dict[str, Any]], - risk_data: List[Dict[str, Any]], - risk_definitions: List[Dict[str, Any]], - alternatives: List[Dict[str, Any]], - alternative_technologies: List[Dict[str, Any]], + metadata: dict[str, Any], + resource_type_mapping: dict[str, Any], + resource_inventory: list[dict[str, Any]], + cost_data: list[dict[str, Any]], + scoring_data: dict[str, Any] | None, + risk_data: list[dict[str, Any]], + risk_definitions: list[dict[str, Any]], + alternatives: list[dict[str, Any]], + alternative_technologies: list[dict[str, Any]], exit_strategy: int, ) -> str: # Define the PDF path diff --git a/core/utils_report_common.py b/core/utils_report_common.py index 61e2dab..cac8368 100644 --- a/core/utils_report_common.py +++ b/core/utils_report_common.py @@ -1,6 +1,6 @@ from collections import defaultdict from datetime import datetime -from typing import Any, Dict, List, Optional, Tuple +from typing import Any CURRENCY_SYMBOLS = { "USD": "$", @@ -9,13 +9,13 @@ } -def sort_cost_data(cost_data: List[Dict[str, Any]]) -> List[Dict[str, Any]]: +def sort_cost_data(cost_data: list[dict[str, Any]]) -> list[dict[str, Any]]: return sorted(cost_data, key=lambda x: datetime.strptime(x["month"], "%Y-%m-%d")) def summarize_costs( - cost_data: List[Dict[str, Any]], *, last_n: Optional[int] = None -) -> Tuple[List[str], List[float], float, str, str]: + cost_data: list[dict[str, Any]], *, last_n: int | None = None +) -> tuple[list[str], list[float], float, str, str]: sorted_costs = sort_cost_data(cost_data) if last_n is not None: sorted_costs = sorted_costs[-last_n:] @@ -37,12 +37,12 @@ def summarize_costs( def summarize_risks( - risk_data: List[Dict[str, Any]], - risk_definitions: List[Dict[str, Any]], + risk_data: list[dict[str, Any]], + risk_definitions: list[dict[str, Any]], *, - resource_name_map: Optional[Dict[str, str]] = None, - resource_id_map: Optional[Dict[str, int]] = None, -) -> Tuple[List[Dict[str, Any]], Dict[str, int]]: + resource_name_map: dict[str, str] | None = None, + resource_id_map: dict[str, int] | None = None, +) -> tuple[list[dict[str, Any]], dict[str, int]]: risk_def_map = {rd["id"]: rd for rd in risk_definitions} severity_counts = {"high": 0, "medium": 0, "low": 0} @@ -115,18 +115,18 @@ def summarize_risks( def summarize_alternative_technologies( - resource_inventory: List[Dict[str, Any]], - alternatives: List[Dict[str, Any]], - alternative_technologies: List[Dict[str, Any]], + resource_inventory: list[dict[str, Any]], + alternatives: list[dict[str, Any]], + alternative_technologies: list[dict[str, Any]], exit_strategy: int, -) -> Dict[str, List[Dict[str, Any]]]: +) -> dict[str, list[dict[str, Any]]]: active_technologies = { tech["id"]: tech for tech in alternative_technologies if tech.get("status") == "t" } - grouped_alt_tech: Dict[str, List[Dict[str, Any]]] = { + grouped_alt_tech: dict[str, list[dict[str, Any]]] = { str(resource["resource_type"]): [] for resource in resource_inventory } @@ -154,11 +154,11 @@ def summarize_alternative_technologies( def enrich_resource_inventory( - resource_inventory: List[Dict[str, Any]], - resource_type_mapping: Dict[str, Dict[str, Any]], + resource_inventory: list[dict[str, Any]], + resource_type_mapping: dict[str, dict[str, Any]], *, - report_path: Optional[str] = None, -) -> List[Dict[str, Any]]: + report_path: str | None = None, +) -> list[dict[str, Any]]: enriched_resources = [] for idx, resource in enumerate(resource_inventory): resource_type = str(resource["resource_type"]) diff --git a/core/utils_report_html.py b/core/utils_report_html.py index 5bd8617..18f20d8 100644 --- a/core/utils_report_html.py +++ b/core/utils_report_html.py @@ -1,6 +1,6 @@ # core/utils_report_html.py import logging -from typing import List, Dict, Any, Tuple +from typing import Any from core.utils_report_common import ( summarize_alternative_technologies, @@ -14,16 +14,16 @@ def transform_cost_inventory_for_html( - cost_data: List[Dict[str, Any]], -) -> Tuple[List[str], List[float], float, str, str]: + cost_data: list[dict[str, Any]], +) -> tuple[list[str], list[float], float, str, str]: return summarize_costs(cost_data) def transform_risk_inventory_for_html( - risk_data: List[Dict[str, Any]], - risk_definitions: List[Dict[str, Any]], - resource_inventory: Dict[str, Dict[str, Any]], -) -> Tuple[List[Dict[str, Any]], Dict[str, int]]: + risk_data: list[dict[str, Any]], + risk_definitions: list[dict[str, Any]], + resource_inventory: dict[str, dict[str, Any]], +) -> tuple[list[dict[str, Any]], dict[str, int]]: severity_order = {"high": 1, "medium": 2, "low": 3} resource_name_map = { str(key): value["name"] for key, value in resource_inventory.items() @@ -38,11 +38,11 @@ def transform_risk_inventory_for_html( def transform_alt_tech_for_html( - resource_inventory: List[Dict[str, Any]], - alternatives: List[Dict[str, Any]], - alternative_technologies: List[Dict[str, Any]], + resource_inventory: list[dict[str, Any]], + alternatives: list[dict[str, Any]], + alternative_technologies: list[dict[str, Any]], exit_strategy: int, -) -> List[Dict[str, Any]]: +) -> list[dict[str, Any]]: alt_tech_data = [] grouped_alt_tech = summarize_alternative_technologies( resource_inventory, diff --git a/core/utils_report_json.py b/core/utils_report_json.py index 8ef847b..c9ac4cd 100644 --- a/core/utils_report_json.py +++ b/core/utils_report_json.py @@ -1,6 +1,6 @@ # core/utils_report_json.py import logging -from typing import List, Dict, Any +from typing import Any from core.utils_report_common import ( enrich_resource_inventory, @@ -15,9 +15,9 @@ def transform_resource_inventory_for_json( - resource_inventory: List[Dict[str, Any]], - resource_type_mapping: Dict[str, Dict[str, Any]], -) -> List[Dict[str, Any]]: + resource_inventory: list[dict[str, Any]], + resource_type_mapping: dict[str, dict[str, Any]], +) -> list[dict[str, Any]]: enriched_resources = enrich_resource_inventory( resource_inventory, resource_type_mapping ) @@ -34,8 +34,8 @@ def transform_resource_inventory_for_json( def transform_cost_inventory_for_json( - cost_data: List[Dict[str, Any]], -) -> List[Dict[str, Any]]: + cost_data: list[dict[str, Any]], +) -> list[dict[str, Any]]: sorted_cost_data = sort_cost_data(cost_data) cost_inventory = [ @@ -50,10 +50,10 @@ def transform_cost_inventory_for_json( def transform_risk_inventory_for_json( - risk_data: List[Dict[str, Any]], - risk_definitions: List[Dict[str, Any]], - resource_inventory: List[Dict[str, Any]], -) -> List[Dict[str, Any]]: + risk_data: list[dict[str, Any]], + risk_definitions: list[dict[str, Any]], + resource_inventory: list[dict[str, Any]], +) -> list[dict[str, Any]]: resource_id_map = { str(value["resource_type"]): key + 1 for key, value in enumerate(resource_inventory) @@ -77,11 +77,11 @@ def transform_risk_inventory_for_json( def transform_alt_tech_for_json( - resource_inventory: List[Dict[str, Any]], - alternatives: List[Dict[str, Any]], - alternative_technologies: List[Dict[str, Any]], + resource_inventory: list[dict[str, Any]], + alternatives: list[dict[str, Any]], + alternative_technologies: list[dict[str, Any]], exit_strategy: int, -) -> Dict[int, List[Dict[str, Any]]]: +) -> dict[int, list[dict[str, Any]]]: resource_id_map = { str(value["resource_type"]): key + 1 for key, value in enumerate(resource_inventory) diff --git a/core/utils_report_pdf.py b/core/utils_report_pdf.py index 63e11d1..12d898e 100644 --- a/core/utils_report_pdf.py +++ b/core/utils_report_pdf.py @@ -3,7 +3,7 @@ import math import logging from datetime import datetime -from typing import List, Dict, Any, Tuple +from typing import Any from math import cos, sin, radians # ReportLab @@ -34,8 +34,8 @@ def transform_resource_inventory_for_pdf( - resource_inventory: list, resource_type_mapping: Dict[str, Any], report_path: str -) -> List[Dict[str, Any]]: + resource_inventory: list, resource_type_mapping: dict[str, Any], report_path: str +) -> list[dict[str, Any]]: enriched_resources = enrich_resource_inventory( resource_inventory, resource_type_mapping, @@ -55,14 +55,14 @@ def transform_resource_inventory_for_pdf( def transform_cost_inventory_for_pdf( cost_data: list, -) -> Tuple[List[str], List[float], str]: +) -> tuple[list[str], list[float], str]: months, costs, _, _, currency_symbol = summarize_costs(cost_data, last_n=6) return months, costs, currency_symbol def transform_risk_inventory_for_pdf( risk_data: list, risk_definitions: list, resource_inventory: list -) -> Tuple[List[Dict[str, Any]], Dict[str, int]]: +) -> tuple[list[dict[str, Any]], dict[str, int]]: risks, severity_counts = summarize_risks(risk_data, risk_definitions) return [ { @@ -76,12 +76,12 @@ def transform_risk_inventory_for_pdf( def transform_alt_tech_for_pdf( resource_inventory: list, - resource_type_mapping: Dict[str, Any], + resource_type_mapping: dict[str, Any], alternatives: list, alternative_technologies: list, exit_strategy: int, report_path: str, -) -> List[Dict[str, Any]]: +) -> list[dict[str, Any]]: grouped_alt_tech = summarize_alternative_technologies( resource_inventory, alternatives, @@ -203,7 +203,7 @@ def draw_header_footer(report_path: str, canvas, doc) -> None: canvas.restoreState() -def draw_risk_chart(risk_chart_data: Dict[str, int]) -> Drawing: +def draw_risk_chart(risk_chart_data: dict[str, int]) -> Drawing: # Define colors for each severity and their border colors severity_colors = { "high": HexColor("#991b1b"), @@ -280,7 +280,7 @@ def draw_risk_chart(risk_chart_data: Dict[str, int]) -> Drawing: return d -def draw_cost_chart(months: List[str], costs: List[float]) -> Drawing: +def draw_cost_chart(months: list[str], costs: list[float]) -> Drawing: # Create a drawing for the bar chart d = Drawing(7.5 * cm, 5 * cm) diff --git a/core/utils_sync.py b/core/utils_sync.py index 500c24d..19548e5 100644 --- a/core/utils_sync.py +++ b/core/utils_sync.py @@ -7,7 +7,7 @@ import time import config import requests -from typing import Any, Dict, List +from typing import Any from core.utils_db import load_data @@ -33,13 +33,13 @@ def _build_payload( exit_strategy: int, cloud_service_provider: int, assessment_type: int, -) -> Dict[str, Any]: +) -> dict[str, Any]: db_path = os.path.join(report_path, "data", "assessment.db") - resource_rows: List[Dict[str, Any]] = load_data( + resource_rows: list[dict[str, Any]] = load_data( "resource_inventory", db_path=db_path ) - cost_rows: List[Dict[str, Any]] = load_data("cost_inventory", db_path=db_path) + cost_rows: list[dict[str, Any]] = load_data("cost_inventory", db_path=db_path) res_payload = [ { @@ -61,7 +61,7 @@ def _build_payload( engine_version = getattr(config, "CLI_VERSION", "v1.0.0").strip() now = int(time.time()) - payload: Dict[str, Any] = { + payload: dict[str, Any] = { "id": os.urandom(16).hex(), "object": "event", "cli_version": engine_version, @@ -89,10 +89,10 @@ def post_assessment( name: str, started_at: int, report_path: str, - meta: Dict[str, int], + meta: dict[str, int], token: str, timeout: int = 10, -) -> Dict[str, Any]: +) -> dict[str, Any]: host = getattr(config, "HOST", "").strip() if not host: return {"success": False, "payload": None, "logs": "HOST missing in config.py"} diff --git a/utils/azure.py b/utils/azure.py index d179394..3564522 100644 --- a/utils/azure.py +++ b/utils/azure.py @@ -2,7 +2,7 @@ import logging import shutil import subprocess -from typing import List, Any +from typing import Any from rich.console import Console from azure.identity import AzureCliCredential from azure.core.exceptions import ClientAuthenticationError @@ -40,7 +40,7 @@ def is_azure_cli_token_expired() -> bool: return False -def select_subscription(subscriptions: List[Any]) -> Any: +def select_subscription(subscriptions: list[Any]) -> Any: # logger.info("Listing available subscriptions for selection.") console.print("Available Subscriptions:") for idx, sub in enumerate(subscriptions, start=1): @@ -58,7 +58,7 @@ def select_subscription(subscriptions: List[Any]) -> Any: console.print(f"[red]{e} Please select a valid number.[/red]") -def select_resource_group(resource_groups: List[Any]) -> str: +def select_resource_group(resource_groups: list[Any]) -> str: # logger.info("Listing available resource groups for selection.") console.print("Available Resource Groups:") for idx, rg in enumerate(resource_groups, start=1): diff --git a/utils/connection.py b/utils/connection.py index 0b71ef6..893e9d1 100644 --- a/utils/connection.py +++ b/utils/connection.py @@ -3,7 +3,6 @@ import logging import requests -from typing import Tuple, Optional logger = logging.getLogger("main.utils.connection") @@ -24,7 +23,7 @@ def _build_url(host: str) -> str: def get_jwt_token( host: str | None = None, key: str | None = None, *, timeout: int = 10 -) -> Optional[str]: +) -> str | None: host = host or getattr(config, "HOST", "") if config else "" key = key or getattr(config, "KEY", "") if config else "" @@ -63,7 +62,7 @@ def get_jwt_token( return None -def resolve_mode() -> Tuple[str, Optional[str]]: +def resolve_mode() -> tuple[str, str | None]: host = getattr(config, "HOST", "") if config else "" key = getattr(config, "KEY", "") if config else "" diff --git a/utils/data.py b/utils/data.py index 238a914..481660d 100644 --- a/utils/data.py +++ b/utils/data.py @@ -5,8 +5,8 @@ import hashlib import time import requests -from typing import Optional -from datetime import datetime, timedelta + +from datetime import datetime, timedelta, timezone from pathlib import Path from requests.exceptions import RequestException, ConnectionError, Timeout @@ -16,7 +16,7 @@ def get_monday_date() -> str: - now = datetime.utcnow() + now = datetime.now(timezone.utc) monday = now - timedelta(days=now.weekday()) if now.weekday() == 0 and now.hour < 8: @@ -66,7 +66,7 @@ def download_file(url: str, destination: str, retries: int = 3, delay: int = 5) def fetch_remote_checksum( checksum_url: str, retries: int = 3, delay: int = 5 -) -> Optional[str]: +) -> str | None: for attempt in range(retries): try: response = requests.get(checksum_url, timeout=10) diff --git a/utils/sync.py b/utils/sync.py index ec19a61..0ca41ce 100644 --- a/utils/sync.py +++ b/utils/sync.py @@ -4,7 +4,7 @@ import logging import requests import config -from typing import Optional, Dict, Any +from typing import Any from utils.auth import get_jwt_token logger = logging.getLogger("main.utils.sync") @@ -20,12 +20,12 @@ def _build_url(host: str) -> str: def submit_assessment( - payload: Dict[str, Any], + payload: dict[str, Any], *, host: str | None = None, key: str | None = None, timeout: int = 10, -) -> Optional[requests.Response]: +) -> requests.Response | None: host = host or getattr(config, "HOST", "") if config else "" if not host: logger.warning("HOST not configured – skipping assessment sync.") diff --git a/utils/utils.py b/utils/utils.py index 70ea0e9..4ec0072 100644 --- a/utils/utils.py +++ b/utils/utils.py @@ -2,7 +2,7 @@ import os import logging import json -from typing import Optional, Tuple, Dict, Any +from typing import Any from rich.console import Console from rich.style import Style from time import sleep @@ -12,7 +12,7 @@ console = Console() -def load_config(file_path: str) -> Optional[Dict[str, Any]]: +def load_config(file_path: str) -> dict[str, Any] | None: try: # logger.info(f"Attempting to load config file from {file_path}") with open(file_path, "r") as f: @@ -25,7 +25,7 @@ def load_config(file_path: str) -> Optional[Dict[str, Any]]: return None -def prompt_required_inputs() -> Tuple[int, int]: +def prompt_required_inputs() -> tuple[int, int]: while True: try: exit_strategy = int( @@ -60,7 +60,7 @@ def prompt_required_inputs() -> Tuple[int, int]: def print_step( - description: str, status: str = "pending", logs: Optional[str] = None + description: str, status: str = "pending", logs: str | None = None ) -> None: # Define styles for statuses ok_style = Style(color="green", bold=True) diff --git a/utils/validate.py b/utils/validate.py index baa62b2..f733116 100644 --- a/utils/validate.py +++ b/utils/validate.py @@ -1,5 +1,5 @@ # utils/validate.py -from typing import Dict, Any +from typing import Any from .constants import REGION_CHOICES, REQUIRED_FIELDS_AZURE, REQUIRED_FIELDS_AWS @@ -9,7 +9,7 @@ def validate_region(region: str) -> None: raise ValueError(f"Invalid AWS region. Choose from: {', '.join(valid_regions)}") -def validate_config(config: Dict[str, Any]) -> bool: +def validate_config(config: dict[str, Any]) -> bool: try: # Cast key values to integers to handle string input gracefully assessment_type = int(config.get("assessmentType", 0))