Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
34 changes: 17 additions & 17 deletions core/engine.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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 = ""

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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)

Expand All @@ -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(
Expand All @@ -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,
Expand Down Expand Up @@ -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.")
Expand Down Expand Up @@ -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")

Expand All @@ -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,
Expand Down
15 changes: 8 additions & 7 deletions core/utils_aws.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand All @@ -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]:
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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
Expand All @@ -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:
Expand Down
8 changes: 4 additions & 4 deletions core/utils_azure.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:
Expand Down Expand Up @@ -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 = {
Expand All @@ -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:
Expand Down
58 changes: 29 additions & 29 deletions core/utils_report.py
Original file line number Diff line number Diff line change
Expand Up @@ -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
Expand Down Expand Up @@ -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:

Expand Down Expand Up @@ -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
Expand Down Expand Up @@ -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
Expand Down
36 changes: 18 additions & 18 deletions core/utils_report_common.py
Original file line number Diff line number Diff line change
@@ -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": "$",
Expand All @@ -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:]
Expand All @@ -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}

Expand Down Expand Up @@ -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
}

Expand Down Expand Up @@ -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"])
Expand Down
Loading