TechDebt: MyPy CostExplorer (#5593)

This commit is contained in:
Bert Blommers 2022-10-23 20:42:14 +00:00 committed by GitHub
parent 6f3b250fc7
commit c0b581aa08
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
5 changed files with 53 additions and 31 deletions

View File

@ -3,7 +3,7 @@ from moto.core.exceptions import JsonRESTError
class CostCategoryNotFound(JsonRESTError):
def __init__(self, ccd_id):
def __init__(self, ccd_id: str):
super().__init__(
"ResourceNotFoundException", f"No Cost Categories found with ID {ccd_id}"
)

View File

@ -5,11 +5,18 @@ from moto.core import BaseBackend, BaseModel
from moto.core.utils import BackendDict
from moto.utilities.tagging_service import TaggingService
from moto.moto_api._internal import mock_random
from typing import Any, Dict, List, Tuple
class CostCategoryDefinition(BaseModel):
def __init__(
self, account_id, name, rule_version, rules, default_value, split_charge_rules
self,
account_id: str,
name: str,
rule_version: str,
rules: List[Dict[str, Any]],
default_value: str,
split_charge_rules: List[Dict[str, Any]],
):
self.name = name
self.rule_version = rule_version
@ -18,13 +25,19 @@ class CostCategoryDefinition(BaseModel):
self.split_charge_rules = split_charge_rules
self.arn = f"arn:aws:ce::{account_id}:costcategory/{str(mock_random.uuid4())}"
def update(self, rule_version, rules, default_value, split_charge_rules):
def update(
self,
rule_version: str,
rules: List[Dict[str, Any]],
default_value: str,
split_charge_rules: List[Dict[str, Any]],
) -> None:
self.rule_version = rule_version
self.rules = rules
self.default_value = default_value
self.split_charge_rules = split_charge_rules
def to_json(self):
def to_json(self) -> Dict[str, Any]:
return {
"CostCategoryArn": self.arn,
"Name": self.name,
@ -38,20 +51,20 @@ class CostCategoryDefinition(BaseModel):
class CostExplorerBackend(BaseBackend):
"""Implementation of CostExplorer APIs."""
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.cost_categories = dict()
self.cost_categories: Dict[str, CostCategoryDefinition] = dict()
self.tagger = TaggingService()
def create_cost_category_definition(
self,
name,
rule_version,
rules,
default_value,
split_charge_rules,
tags,
):
name: str,
rule_version: str,
rules: List[Dict[str, Any]],
default_value: str,
split_charge_rules: List[Dict[str, Any]],
tags: List[Dict[str, str]],
) -> Tuple[str, str]:
"""
The EffectiveOn and ResourceTags-parameters are not yet implemented
"""
@ -67,7 +80,9 @@ class CostExplorerBackend(BaseBackend):
self.tag_resource(ccd.arn, tags)
return ccd.arn, ""
def describe_cost_category_definition(self, cost_category_arn):
def describe_cost_category_definition(
self, cost_category_arn: str
) -> CostCategoryDefinition:
"""
The EffectiveOn-parameter is not yet implemented
"""
@ -76,7 +91,9 @@ class CostExplorerBackend(BaseBackend):
raise CostCategoryNotFound(ccd_id)
return self.cost_categories[cost_category_arn]
def delete_cost_category_definition(self, cost_category_arn):
def delete_cost_category_definition(
self, cost_category_arn: str
) -> Tuple[str, str]:
"""
The EffectiveOn-parameter is not yet implemented
"""
@ -84,8 +101,13 @@ class CostExplorerBackend(BaseBackend):
return cost_category_arn, ""
def update_cost_category_definition(
self, cost_category_arn, rule_version, rules, default_value, split_charge_rules
):
self,
cost_category_arn: str,
rule_version: str,
rules: List[Dict[str, Any]],
default_value: str,
split_charge_rules: List[Dict[str, Any]],
) -> Tuple[str, str]:
"""
The EffectiveOn-parameter is not yet implemented
"""
@ -94,13 +116,13 @@ class CostExplorerBackend(BaseBackend):
return cost_category_arn, ""
def list_tags_for_resource(self, resource_arn):
def list_tags_for_resource(self, resource_arn: str) -> List[Dict[str, str]]:
return self.tagger.list_tags_for_resource(arn=resource_arn)["Tags"]
def tag_resource(self, resource_arn, tags):
def tag_resource(self, resource_arn: str, tags: List[Dict[str, str]]) -> None:
self.tagger.tag_resource(resource_arn, tags)
def untag_resource(self, resource_arn, tag_keys):
def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
self.tagger.untag_resource_using_names(resource_arn, tag_keys)

View File

@ -2,18 +2,18 @@
import json
from moto.core.responses import BaseResponse
from .models import ce_backends
from .models import ce_backends, CostExplorerBackend
class CostExplorerResponse(BaseResponse):
"""Handler for CostExplorer requests and responses."""
@property
def ce_backend(self):
def ce_backend(self) -> CostExplorerBackend:
"""Return backend instance specific for this region."""
return ce_backends[self.current_account]["global"]
def create_cost_category_definition(self):
def create_cost_category_definition(self) -> str:
params = json.loads(self.body)
name = params.get("Name")
rule_version = params.get("RuleVersion")
@ -36,7 +36,7 @@ class CostExplorerResponse(BaseResponse):
dict(CostCategoryArn=cost_category_arn, EffectiveStart=effective_start)
)
def describe_cost_category_definition(self):
def describe_cost_category_definition(self) -> str:
params = json.loads(self.body)
cost_category_arn = params.get("CostCategoryArn")
cost_category = self.ce_backend.describe_cost_category_definition(
@ -44,7 +44,7 @@ class CostExplorerResponse(BaseResponse):
)
return json.dumps(dict(CostCategory=cost_category.to_json()))
def delete_cost_category_definition(self):
def delete_cost_category_definition(self) -> str:
params = json.loads(self.body)
cost_category_arn = params.get("CostCategoryArn")
(
@ -57,7 +57,7 @@ class CostExplorerResponse(BaseResponse):
dict(CostCategoryArn=cost_category_arn, EffectiveEnd=effective_end)
)
def update_cost_category_definition(self):
def update_cost_category_definition(self) -> str:
params = json.loads(self.body)
cost_category_arn = params.get("CostCategoryArn")
rule_version = params.get("RuleVersion")
@ -78,20 +78,20 @@ class CostExplorerResponse(BaseResponse):
dict(CostCategoryArn=cost_category_arn, EffectiveStart=effective_start)
)
def list_tags_for_resource(self):
def list_tags_for_resource(self) -> str:
params = json.loads(self.body)
resource_arn = params.get("ResourceArn")
tags = self.ce_backend.list_tags_for_resource(resource_arn)
return json.dumps({"ResourceTags": tags})
def tag_resource(self):
def tag_resource(self) -> str:
params = json.loads(self.body)
resource_arn = params.get("ResourceArn")
tags = params.get("ResourceTags")
self.ce_backend.tag_resource(resource_arn, tags)
return json.dumps({})
def untag_resource(self):
def untag_resource(self) -> str:
params = json.loads(self.body)
resource_arn = params.get("ResourceArn")
tag_names = params.get("ResourceTagKeys")

View File

@ -22,7 +22,7 @@ class TaggingService:
result[key] = val
return result
def list_tags_for_resource(self, arn: str) -> List[Dict[str, str]]:
def list_tags_for_resource(self, arn: str) -> Dict[str, List[Dict[str, str]]]:
"""Return list of tags inside dict with key of "tag_name".
Useful for describe functions; this return value can be added to

View File

@ -18,7 +18,7 @@ disable = W,C,R,E
enable = anomalous-backslash-in-string, arguments-renamed, dangerous-default-value, deprecated-module, function-redefined, import-self, redefined-builtin, redefined-outer-name, reimported, pointless-statement, super-with-arguments, unused-argument, unused-import, unused-variable, useless-else-on-loop, wildcard-import
[mypy]
files= moto/a*,moto/b*
files= moto/a*,moto/b*,moto/ce
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract