Skip to content
Open
Show file tree
Hide file tree
Changes from 4 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
7 changes: 4 additions & 3 deletions src/tirith/providers/infracost/handler.py
Original file line number Diff line number Diff line change
@@ -1,9 +1,10 @@
import logging
from typing import List, Dict, Any

logger = logging.getLogger(__name__)


def __get_all_costs(operation_type, input_data):
def __get_all_costs(operation_type: str, input_data: dict) -> float:
logger.debug(f"costType : {operation_type}")
pointer = {
"total_monthly_cost": ["totalMonthlyCost", "monthlyCost"],
Expand Down Expand Up @@ -38,7 +39,7 @@ def __get_all_costs(operation_type, input_data):
raise KeyError("projects not found in input_data")


def __get_resources_costs(resource_type, operation_type, input_data):
def __get_resources_costs(resource_type: str, operation_type: str, input_data: dict) -> float:
logger.debug(f"costType : {operation_type}")
pointer = {"total_monthly_cost": "totalMonthlyCost", "total_hourly_cost": "totalHourlyCost"}
pointer = {
Expand Down Expand Up @@ -77,7 +78,7 @@ def __get_resources_costs(resource_type, operation_type, input_data):
raise KeyError("projects not found in input_data")


def provide(provider_args, input_data):
def provide(provider_args: dict, input_data: dict) -> List[Dict[str, Any]]:
logger.debug("infracost provider")
logger.debug(f"infracost provider inputs : {provider_args}")
try:
Expand Down
13 changes: 7 additions & 6 deletions src/tirith/providers/json/handler.py
Original file line number Diff line number Diff line change
@@ -1,15 +1,15 @@
import pydash

from typing import Callable, Dict, List
from typing import Callable, Dict, List, Any
from ..common import create_result_dict, ProviderError, get_path_value_from_dict


class PydashPathNotFound:
pass


def _get_path_value_from_dict(splitted_paths, input_dict):
final_data = []
def _get_path_value_from_dict(splitted_paths: List[str], input_dict: Dict[str, Any]) -> List[Any]:
final_data: List[Any] = []
for i, expression in enumerate(splitted_paths):
intermediate_val = pydash.get(input_dict, expression, default=PydashPathNotFound)
if isinstance(intermediate_val, list) and i < len(splitted_paths) - 1:
Expand All @@ -28,7 +28,7 @@ def _get_path_value_from_dict(splitted_paths, input_dict):
return final_data


def get_value(provider_args: Dict, input_data: Dict) -> List[dict]:
def get_value(provider_args: Dict[str, Any], input_data: Dict[str, Any]) -> List[Dict[str, Any]]:
# Must be validated first whether the provider args are valid for this op type
key_path: str = provider_args["key_path"]

Expand All @@ -48,10 +48,11 @@ def get_value(provider_args: Dict, input_data: Dict) -> List[dict]:
return outputs


SUPPORTED_OPS: Dict[str, Callable] = {"get_value": get_value}
SUPPORTED_OPS: Dict[str, Callable[[Dict[str, Any], Dict[str, Any]], List[Dict[str, Any]]]]
Comment thread
kushal9897 marked this conversation as resolved.
Outdated


def provide(provider_args: Dict, input_data: Dict) -> List[Dict]:
def provide(provider_args: Dict[str, Any], input_data: Dict[str, Any]) -> List[Dict[str, Any]]:

operation_type = provider_args["operation_type"]

op_handler = SUPPORTED_OPS.get(operation_type)
Expand Down
7 changes: 4 additions & 3 deletions src/tirith/providers/sg_workflow/handler.py
Original file line number Diff line number Diff line change
@@ -1,12 +1,13 @@
import logging
from typing import Any, Dict, List, Union


logger = logging.getLogger(__name__)


def __getValue(key, data):
def __getValue(key: str, data: Dict[str, any]) -> Union[str, List[str]]:
logger.debug(f"Searching {key} in input data")
result = ""
result: Union[str, List[str]] = ""
if key == "integrationId":
temp = []
if "DeploymentPlatformConfig" in data:
Expand Down Expand Up @@ -70,7 +71,7 @@ def __getValue(key, data):
return result


def provide(provider_args, input_data):
def provide(provider_args: Dict[str, Any], input_data: Dict[str, Any]) -> List[Dict[str, Any]]:
logger.debug("sg_workflow provider")
logger.debug(f"sg_workflow provider inputs : {provider_args}")
try:
Expand Down