diff --git a/moto/backends.py b/moto/backends.py index 2680e73c4..c00331cf5 100644 --- a/moto/backends.py +++ b/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])()) diff --git a/moto/budgets/exceptions.py b/moto/budgets/exceptions.py index e16859e14..03d073383 100644 --- a/moto/budgets/exceptions.py +++ b/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): - super().__init__(__class__.__name__, message) + def __init__(self, message: str): + 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", diff --git a/moto/budgets/models.py b/moto/budgets/models.py index a0841445e..89e010b08 100644 --- a/moto/budgets/models.py +++ b/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 = defaultdict(dict) + self.budgets: Dict[str, Dict[str, Budget]] = 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 """ diff --git a/moto/budgets/responses.py b/moto/budgets/responses.py index c60e92aaa..dcfe42c1c 100644 --- a/moto/budgets/responses.py +++ b/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( diff --git a/setup.cfg b/setup.cfg index a70a36960..e42e449bf 100644 --- a/setup.cfg +++ b/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