TechDebt - MyPy Budgets (#5591)

moto/backends.py

@@ -1,6 +1,8 @@
 import importlib
 import moto
 import sys
+from moto.core.utils import BackendDict
+from typing import Iterable, Tuple
 
 
 decorators = [d for d in dir(moto) if d.startswith("mock_") and not d == "mock_all"]
@@ -12,44 +14,35 @@ BACKENDS["instance_metadata"] = ("instance_metadata", "instance_metadata_backend
 BACKENDS["s3bucket_path"] = ("s3", "s3_backends")
 
 
-def _import_backend(module_name, backends_name):
+def _import_backend(module_name: str, backends_name: str) -> BackendDict:
     module = importlib.import_module("moto." + module_name)
     return getattr(module, backends_name)
 
 
-def backends():
+def backends() -> Iterable[BackendDict]:
     for module_name, backends_name in BACKENDS.values():
         yield _import_backend(module_name, backends_name)
 
 
-def service_backends():
+def service_backends() -> Iterable[BackendDict]:
     services = [(f.name, f.backend) for f in decorator_functions]
     for module_name, backends_name in sorted(set(services)):
         yield _import_backend(module_name, backends_name)
 
 
-def loaded_backends():
+def loaded_backends() -> Iterable[Tuple[str, BackendDict]]:
     loaded_modules = sys.modules.keys()
-    loaded_modules = [m for m in loaded_modules if m.startswith("moto.")]
+    moto_modules = [m for m in loaded_modules if m.startswith("moto.")]
     imported_backends = [
         name
         for name, (module_name, _) in BACKENDS.items()
-        if f"moto.{module_name}" in loaded_modules
+        if f"moto.{module_name}" in moto_modules
     ]
     for name in imported_backends:
         module_name, backends_name = BACKENDS[name]
         yield name, _import_backend(module_name, backends_name)
 
 
-def get_backend(name):
+def get_backend(name: str) -> BackendDict:
     module_name, backends_name = BACKENDS[name]
     return _import_backend(module_name, backends_name)
-
-
-def get_model(name, region_name):
-    for backends_ in backends():
-        for region, backend in backends_.items():
-            if region == region_name:
-                models = getattr(backend.__class__, "__models__", {})
-                if name in models:
-                    return list(getattr(backend, models[name])())

moto/budgets/exceptions.py

@@ -5,9 +5,9 @@ from moto.core.exceptions import JsonRESTError
 class DuplicateRecordException(JsonRESTError):
     code = 400
 
-    def __init__(self, record_type, record_name):
+    def __init__(self, record_type: str, record_name: str):
         super().__init__(
-            __class__.__name__,
+            __class__.__name__,  # type: ignore[name-defined]
             f"Error creating {record_type}: {record_name} - the {record_type} already exists.",
         )
 
@@ -15,14 +15,14 @@ class DuplicateRecordException(JsonRESTError):
 class NotFoundException(JsonRESTError):
     code = 400
 
-    def __init__(self, message):
+    def __init__(self, message: str):
-        super().__init__(__class__.__name__, message)
+        super().__init__(__class__.__name__, message)  # type: ignore[name-defined]
 
 
 class BudgetMissingLimit(JsonRESTError):
     code = 400
 
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__(
             "InvalidParameterException",
             "Unable to create/update budget - please provide one of the followings: Budget Limit/ Planned Budget Limit/ Auto Adjust Data",

moto/budgets/models.py

@@ -3,18 +3,18 @@ from copy import deepcopy
 from datetime import datetime
 from moto.core import BaseBackend, BaseModel
 from moto.core.utils import unix_time, BackendDict
-
+from typing import Any, Dict, Iterable, List
 from .exceptions import BudgetMissingLimit, DuplicateRecordException, NotFoundException
 
 
 class Notification(BaseModel):
-    def __init__(self, details, subscribers):
+    def __init__(self, details: Dict[str, Any], subscribers: Dict[str, Any]):
         self.details = details
         self.subscribers = subscribers
 
 
 class Budget(BaseModel):
-    def __init__(self, budget, notifications):
+    def __init__(self, budget: Dict[str, Any], notifications: List[Dict[str, Any]]):
         if "BudgetLimit" not in budget and "PlannedBudgetLimits" not in budget:
             raise BudgetMissingLimit()
         # Storing the budget as a Dict for now - if we need more control, we can always read/write it back
@@ -33,7 +33,7 @@ class Budget(BaseModel):
                 "End": 3706473600,  # "2087-06-15T00:00:00+00:00"
             }
 
-    def to_dict(self):
+    def to_dict(self) -> Dict[str, Any]:
         cp = deepcopy(self.budget)
         if "CalculatedSpend" not in cp:
             cp["CalculatedSpend"] = {
@@ -56,25 +56,31 @@ class Budget(BaseModel):
             }
         return cp
 
-    def add_notification(self, details, subscribers):
+    def add_notification(
+        self, details: Dict[str, Any], subscribers: Dict[str, Any]
+    ) -> None:
         self.notifications.append(Notification(details, subscribers))
 
-    def delete_notification(self, details):
+    def delete_notification(self, details: Dict[str, Any]) -> None:
         self.notifications = [n for n in self.notifications if n.details != details]
 
-    def get_notifications(self):
+    def get_notifications(self) -> Iterable[Dict[str, Any]]:
         return [n.details for n in self.notifications]
 
 
 class BudgetsBackend(BaseBackend):
     """Implementation of Budgets APIs."""
 
-    def __init__(self, region_name, account_id):
+    def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
-        # {"account_id": {"budget_name": Budget}}
+        self.budgets: Dict[str, Dict[str, Budget]] = defaultdict(dict)
-        self.budgets = defaultdict(dict)
 
-    def create_budget(self, account_id, budget, notifications):
+    def create_budget(
+        self,
+        account_id: str,
+        budget: Dict[str, Any],
+        notifications: List[Dict[str, Any]],
+    ) -> None:
         budget_name = budget["BudgetName"]
         if budget_name in self.budgets[account_id]:
             raise DuplicateRecordException(
@@ -82,26 +88,32 @@ class BudgetsBackend(BaseBackend):
             )
         self.budgets[account_id][budget_name] = Budget(budget, notifications)
 
-    def describe_budget(self, account_id, budget_name):
+    def describe_budget(self, account_id: str, budget_name: str) -> Dict[str, Any]:
         if budget_name not in self.budgets[account_id]:
             raise NotFoundException(
                 f"Unable to get budget: {budget_name} - the budget doesn't exist."
             )
         return self.budgets[account_id][budget_name].to_dict()
 
-    def describe_budgets(self, account_id):
+    def describe_budgets(self, account_id: str) -> Iterable[Dict[str, Any]]:
         """
         Pagination is not yet implemented
         """
         return [budget.to_dict() for budget in self.budgets[account_id].values()]
 
-    def delete_budget(self, account_id, budget_name):
+    def delete_budget(self, account_id: str, budget_name: str) -> None:
         if budget_name not in self.budgets[account_id]:
             msg = f"Unable to delete budget: {budget_name} - the budget doesn't exist. Try creating it first. "
             raise NotFoundException(msg)
         self.budgets[account_id].pop(budget_name)
 
-    def create_notification(self, account_id, budget_name, notification, subscribers):
+    def create_notification(
+        self,
+        account_id: str,
+        budget_name: str,
+        notification: Dict[str, Any],
+        subscribers: Dict[str, Any],
+    ) -> None:
         if budget_name not in self.budgets[account_id]:
             raise NotFoundException(
                 "Unable to create notification - the budget doesn't exist."
@@ -110,14 +122,18 @@ class BudgetsBackend(BaseBackend):
             details=notification, subscribers=subscribers
         )
 
-    def delete_notification(self, account_id, budget_name, notification):
+    def delete_notification(
+        self, account_id: str, budget_name: str, notification: Dict[str, Any]
+    ) -> None:
         if budget_name not in self.budgets[account_id]:
             raise NotFoundException(
                 "Unable to delete notification - the budget doesn't exist."
             )
         self.budgets[account_id][budget_name].delete_notification(details=notification)
 
-    def describe_notifications_for_budget(self, account_id, budget_name):
+    def describe_notifications_for_budget(
+        self, account_id: str, budget_name: str
+    ) -> Iterable[Dict[str, Any]]:
         """
         Pagination has not yet been implemented
         """

moto/budgets/responses.py

@@ -1,18 +1,18 @@
 import json
 
 from moto.core.responses import BaseResponse
-from .models import budgets_backends
+from .models import budgets_backends, BudgetsBackend
 
 
 class BudgetsResponse(BaseResponse):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__(service_name="budgets")
 
     @property
-    def backend(self):
+    def backend(self) -> BudgetsBackend:
         return budgets_backends[self.current_account]["global"]
 
-    def create_budget(self):
+    def create_budget(self) -> str:
         account_id = self._get_param("AccountId")
         budget = self._get_param("Budget")
         notifications = self._get_param("NotificationsWithSubscribers", [])
@@ -21,7 +21,7 @@ class BudgetsResponse(BaseResponse):
         )
         return json.dumps(dict())
 
-    def describe_budget(self):
+    def describe_budget(self) -> str:
         account_id = self._get_param("AccountId")
         budget_name = self._get_param("BudgetName")
         budget = self.backend.describe_budget(
@@ -29,18 +29,18 @@ class BudgetsResponse(BaseResponse):
         )
         return json.dumps(dict(Budget=budget))
 
-    def describe_budgets(self):
+    def describe_budgets(self) -> str:
         account_id = self._get_param("AccountId")
         budgets = self.backend.describe_budgets(account_id=account_id)
         return json.dumps(dict(Budgets=budgets, nextToken=None))
 
-    def delete_budget(self):
+    def delete_budget(self) -> str:
         account_id = self._get_param("AccountId")
         budget_name = self._get_param("BudgetName")
         self.backend.delete_budget(account_id=account_id, budget_name=budget_name)
         return json.dumps(dict())
 
-    def create_notification(self):
+    def create_notification(self) -> str:
         account_id = self._get_param("AccountId")
         budget_name = self._get_param("BudgetName")
         notification = self._get_param("Notification")
@@ -53,7 +53,7 @@ class BudgetsResponse(BaseResponse):
         )
         return json.dumps(dict())
 
-    def delete_notification(self):
+    def delete_notification(self) -> str:
         account_id = self._get_param("AccountId")
         budget_name = self._get_param("BudgetName")
         notification = self._get_param("Notification")
@@ -62,7 +62,7 @@ class BudgetsResponse(BaseResponse):
         )
         return json.dumps(dict())
 
-    def describe_notifications_for_budget(self):
+    def describe_notifications_for_budget(self) -> str:
         account_id = self._get_param("AccountId")
         budget_name = self._get_param("BudgetName")
         notifications = self.backend.describe_notifications_for_budget(

setup.cfg

@@ -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*
+files= moto/a*,moto/budgets
 show_column_numbers=True
 show_error_codes = True
 disable_error_code=abstract