Techdebt: typing for backends (#7069)

moto/awslambda/models.py

@@ -1281,7 +1281,7 @@ def update_from_cloudformation_json(  # type: ignore[misc]
         properties = cloudformation_json["Properties"]
         event_source_uuid = original_resource.uuid
         lambda_backend = lambda_backends[account_id][region_name]
-        return lambda_backend.update_event_source_mapping(event_source_uuid, properties)
+        return lambda_backend.update_event_source_mapping(event_source_uuid, properties)  # type: ignore[return-value]
 
     @classmethod
     def delete_from_cloudformation_json(  # type: ignore[misc]
@@ -1331,7 +1331,7 @@ def create_from_cloudformation_json(  # type: ignore[misc]
         properties = cloudformation_json["Properties"]
         function_name = properties["FunctionName"]
         func = lambda_backends[account_id][region_name].publish_function(function_name)
-        spec = {"Version": func.version}
+        spec = {"Version": func.version}  # type: ignore[union-attr]
         return LambdaVersion(spec)
 
 

moto/awslambda_simple/responses.py

@@ -5,4 +5,4 @@
 class LambdaSimpleResponse(LambdaResponse):
     @property
     def backend(self) -> LambdaBackend:
-        return lambda_simple_backends[self.current_account][self.region]
+        return lambda_simple_backends[self.current_account][self.region]  # type: ignore[return-value]

moto/batch/models.py

@@ -211,7 +211,7 @@ def create_from_cloudformation_json(  # type: ignore[misc]
             priority=properties["Priority"],
             state=properties.get("State", "ENABLED"),
             compute_env_order=compute_envs,
-            schedule_policy={},
+            schedule_policy=None,
         )
 
 
@@ -448,18 +448,18 @@ def create_from_cloudformation_json(  # type: ignore[misc]
             tags=lowercase_first_key(properties.get("Tags", {})),
             retry_strategy=lowercase_first_key(properties["RetryStrategy"]),
             container_properties=(
-                lowercase_first_key(properties["ContainerProperties"])
+                lowercase_first_key(properties["ContainerProperties"])  # type: ignore[arg-type]
                 if "ContainerProperties" in properties
                 else None
             ),
             node_properties=(
-                lowercase_first_key(properties["NodeProperties"])
+                lowercase_first_key(properties["NodeProperties"])  # type: ignore[arg-type]
                 if "NodeProperties" in properties
                 else None
             ),
             timeout=lowercase_first_key(properties.get("timeout", {})),
-            platform_capabilities=None,
-            propagate_tags=None,
+            platform_capabilities=None,  # type: ignore[arg-type]
+            propagate_tags=None,  # type: ignore[arg-type]
         )
 
 

moto/batch_simple/responses.py

@@ -9,4 +9,4 @@ def batch_backend(self) -> BatchBackend:
         :return: Batch Backend
         :rtype: moto.batch.models.BatchBackend
         """
-        return batch_simple_backends[self.current_account][self.region]
+        return batch_simple_backends[self.current_account][self.region]  # type: ignore[return-value]

moto/cloudformation/parsing.py

@@ -598,9 +598,10 @@ def transform_mapping(self) -> None:
                     location = params["Location"]
                     bucket_name, name = bucket_and_name_from_url(location)
                     key = s3_backends[self._account_id]["global"].get_object(
-                        bucket_name, name
+                        bucket_name,  # type: ignore[arg-type]
+                        name,
                     )
-                    self._parsed_resources.update(json.loads(key.value))
+                    self._parsed_resources.update(json.loads(key.value))  # type: ignore[union-attr]
 
     def parse_ssm_parameter(self, value: str, value_type: str) -> str:
         # The Value in SSM parameters is the SSM parameter path
@@ -609,9 +610,9 @@ def parse_ssm_parameter(self, value: str, value_type: str) -> str:
         parameter = ssm_backends[self._account_id][self._region_name].get_parameter(
             value
         )
-        actual_value = parameter.value
+        actual_value = parameter.value  # type: ignore[union-attr]
         if value_type.find("List") > 0:
-            return actual_value.split(",")
+            return actual_value.split(",")  # type: ignore[return-value]
         return actual_value
 
     def load_parameters(self) -> None:

moto/cloudformation/utils.py

@@ -109,4 +109,4 @@ def get_stack_from_s3_url(template_url: str, account_id: str) -> str:
             key_name = template_url_parts.path.lstrip("/")
 
     key = s3_backends[account_id]["global"].get_object(bucket_name, key_name)
-    return key.value.decode("utf-8")
+    return key.value.decode("utf-8")  # type: ignore[union-attr]

moto/cognitoidp/responses.py

@@ -469,7 +469,7 @@ def forgot_password(self) -> str:
         ].forgot_password(client_id, username)
         self.response_headers[
             "x-moto-forgot-password-confirmation-code"
-        ] = confirmation_code
+        ] = confirmation_code  # type: ignore[assignment]
         return json.dumps(response)
 
     # This endpoint receives no authorization header, so if moto-server is listening

moto/core/__init__.py

@@ -1,5 +1,5 @@
 from .models import DEFAULT_ACCOUNT_ID  # noqa
-from .base_backend import BaseBackend, BackendDict  # noqa
+from .base_backend import BaseBackend, BackendDict, SERVICE_BACKEND  # noqa
 from .common_models import BaseModel  # noqa
 from .common_models import CloudFormationModel, CloudWatchMetricProvider  # noqa
 from .models import patch_client, patch_resource  # noqa

moto/core/base_backend.py

@@ -2,14 +2,15 @@
 import string
 from functools import lru_cache
 from threading import RLock
-from typing import Any, ClassVar, Dict, Iterator, List, Optional, TypeVar
+from typing import Any, Callable, ClassVar, Dict, Iterator, List, Optional, TypeVar
 from uuid import uuid4
 
 from boto3 import Session
 
 from moto.settings import allow_unknown_region, enable_iso_regions
 
 from .model_instances import model_data
+from .responses import TYPE_RESPONSE
 from .utils import convert_regex_to_flask_path
 
 
@@ -47,7 +48,7 @@ def _url_module(self) -> Any:  # type: ignore[misc]
         return backend_urls_module
 
     @property
-    def urls(self) -> Dict[str, str]:
+    def urls(self) -> Dict[str, Callable[[Any, str, Any], TYPE_RESPONSE]]:  # type: ignore[misc]
         """
         A dictionary of the urls to be mocked with this service and the handlers
         that should be called in their place
@@ -81,7 +82,7 @@ def urls(self) -> Dict[str, str]:
         return urls
 
     @property
-    def url_paths(self) -> Dict[str, str]:
+    def url_paths(self) -> Dict[str, Callable[[Any, str, Any], TYPE_RESPONSE]]:  # type: ignore[misc]
         """
         A dictionary of the paths of the urls to be mocked with this service and
         the handlers that should be called in their place
@@ -103,7 +104,7 @@ def url_bases(self) -> List[str]:
         return self._url_module.url_bases
 
     @property
-    def flask_paths(self) -> Dict[str, str]:
+    def flask_paths(self) -> Dict[str, Callable[[Any, str, Any], TYPE_RESPONSE]]:  # type: ignore[misc]
         """
         The url paths that will be used for the flask server
         """
@@ -259,7 +260,7 @@ def __getitem__(self, region_name: str) -> SERVICE_BACKEND:
         return super().__getitem__(region_name)
 
 
-class BackendDict(Dict[str, AccountSpecificBackend]):  # type: ignore[type-arg]
+class BackendDict(Dict[str, AccountSpecificBackend[SERVICE_BACKEND]]):
     """
     Data Structure to store everything related to a specific service.
     Format:
@@ -269,7 +270,7 @@ class BackendDict(Dict[str, AccountSpecificBackend]):  # type: ignore[type-arg]
 
     def __init__(
         self,
-        backend: Any,
+        backend: "type[SERVICE_BACKEND]",
         service_name: str,
         use_boto3_regions: bool = True,
         additional_regions: Optional[List[str]] = None,
@@ -292,7 +293,7 @@ def __ne__(self, other: Any) -> bool:
         return not self.__eq__(other)
 
     @lru_cache()
-    def __getitem__(self, account_id: str) -> AccountSpecificBackend:  # type: ignore
+    def __getitem__(self, account_id: str) -> AccountSpecificBackend[SERVICE_BACKEND]:
         self._create_account_specific_backend(account_id)
         return super().__getitem__(account_id)
 
@@ -305,7 +306,11 @@ def __iter__(self) -> Iterator[str]:
     def __len__(self) -> int:
         return super().__len__()
 
-    def __setitem__(self, key: str, value: AccountSpecificBackend) -> None:  # type: ignore[type-arg]
+    def __setitem__(
+        self,
+        key: str,
+        value: AccountSpecificBackend[SERVICE_BACKEND],
+    ) -> None:
         super().__setitem__(key, value)
 
     def _create_account_specific_backend(self, account_id: str) -> None:

moto/core/models.py

@@ -27,7 +27,7 @@
 
 from moto import settings
 
-from .base_backend import BackendDict
+from .base_backend import SERVICE_BACKEND, BackendDict, BaseBackend
 from .botocore_stubber import BotocoreStubber
 from .custom_responses_mock import (
     CallbackResponse,
@@ -53,13 +53,13 @@ class BaseMockAWS(ContextManager["BaseMockAWS"]):
     nested_count = 0
     mocks_active = False
 
-    def __init__(self, backends: BackendDict):
+    def __init__(self, backends: BackendDict[SERVICE_BACKEND]):
         from moto.instance_metadata import instance_metadata_backends
         from moto.moto_api._internal.models import moto_api_backend
 
         self.backends = backends
 
-        self.backends_for_urls = []
+        self.backends_for_urls: list[BaseBackend] = []
         default_account_id = DEFAULT_ACCOUNT_ID
         default_backends = [
             instance_metadata_backends[default_account_id]["global"],
@@ -534,7 +534,7 @@ def disable_patching(self) -> None:
 class base_decorator:
     mock_backend = MockAWS
 
-    def __init__(self, backends: BackendDict):
+    def __init__(self, backends: BackendDict[SERVICE_BACKEND]):
         self.backends = backends
 
     @overload

moto/ds/models.py

@@ -133,13 +133,13 @@ def create_eni(
                 self.region
             ].create_network_interface(
                 subnet=subnet_id,
-                private_ip_address=None,
+                private_ip_address=None,  # type: ignore[arg-type]
                 group_ids=[security_group_id],
                 description=f"AWS created network interface for {self.directory_id}",
             )
             eni_ids.append(eni_info.id)
             subnet_ips.append(eni_info.private_ip_address)
-        return eni_ids, subnet_ips
+        return eni_ids, subnet_ips  # type: ignore[return-value]
 
     def delete_eni(self) -> None:
         """Delete ENI for each subnet and the security group."""

moto/ec2/models/__init__.py

@@ -73,7 +73,7 @@ def enable_ebs_encryption_by_default(self) -> None:
         ec2_backend = ec2_backends[self.account_id][self.region_name]  # type: ignore[attr-defined]
         ec2_backend.ebs_encryption_by_default = True
 
-    def get_ebs_encryption_by_default(self) -> None:
+    def get_ebs_encryption_by_default(self) -> bool:
         ec2_backend = ec2_backends[self.account_id][self.region_name]  # type: ignore[attr-defined]
         return ec2_backend.ebs_encryption_by_default
 

moto/ec2/models/elastic_block_store.py

@@ -96,12 +96,11 @@ def create_from_cloudformation_json(  # type: ignore[misc]
         volume_id = properties["VolumeId"]
 
         ec2_backend = ec2_backends[account_id][region_name]
-        attachment = ec2_backend.attach_volume(
+        return ec2_backend.attach_volume(  # type: ignore[return-value]
             volume_id=volume_id,
             instance_id=instance_id,
             device_path=properties["Device"],
         )
-        return attachment
 
 
 class Volume(TaggedEC2Resource, CloudFormationModel):

moto/ec2/models/instances.py

@@ -296,7 +296,7 @@ def create_from_cloudformation_json(  # type: ignore[misc]
         ec2_backend = ec2_backends[account_id][region_name]
         security_group_ids = properties.get("SecurityGroups", [])
         group_names = [
-            ec2_backend.get_security_group_from_id(group_id).name
+            ec2_backend.get_security_group_from_id(group_id).name  # type: ignore[union-attr]
             for group_id in security_group_ids
         ]
 

moto/ec2/models/subnets.py

@@ -1,12 +1,13 @@
 import ipaddress
 import itertools
 from collections import defaultdict
-from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Set, Tuple
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple
 
 from moto.core import CloudFormationModel
 
 if TYPE_CHECKING:
     from moto.ec2.models.instances import Instance
+
 from moto.ec2.models.availability_zones_and_regions import Zone
 
 from ..exceptions import (
@@ -412,7 +413,7 @@ def create_subnet(
 
     def describe_subnets(
         self, subnet_ids: Optional[List[str]] = None, filters: Optional[Any] = None
-    ) -> Iterable[Subnet]:
+    ) -> List[Subnet]:
         # Extract a list of all subnets
         matches = list(
             itertools.chain(*[x.copy().values() for x in self.subnets.copy().values()])

moto/elbv2/models.py

@@ -421,7 +421,7 @@ def update_from_cloudformation_json(  # type: ignore[misc]
         listener_rule = elbv2_backend.modify_rule(
             original_resource.arn, conditions, actions
         )
-        return listener_rule
+        return listener_rule  # type: ignore[return-value]
 
 
 class FakeRule(BaseModel):

moto/events/models.py

@@ -187,8 +187,8 @@ def _send_to_events_archive(self, resource_id: str, event: Dict[str, Any]) -> No
         archive = events_backends[self.account_id][self.region_name].archives.get(
             archive_name
         )
-        if archive.uuid == archive_uuid:
-            archive.events.append(event)
+        if archive.uuid == archive_uuid:  # type: ignore[union-attr]
+            archive.events.append(event)  # type: ignore[arg-type,union-attr]
 
     def _find_api_destination(self, resource_id: str) -> "Destination":
         backend: "EventsBackend" = events_backends[self.account_id][self.region_name]

moto/events/notifications.py

@@ -53,7 +53,7 @@ def _send_safe_notification(
 
             for target in applicable_targets:
                 if target.get("Arn", "").startswith("arn:aws:lambda"):
-                    _invoke_lambda(account_id, target.get("Arn"), event=event)
+                    _invoke_lambda(account_id, target.get("Arn"), event=event)  # type: ignore[arg-type]
 
 
 def _invoke_lambda(account_id: str, fn_arn: str, event: Any) -> None:
@@ -64,7 +64,7 @@ def _invoke_lambda(account_id: str, fn_arn: str, event: Any) -> None:
     body = json.dumps(event)
     lambda_backends[account_id][lmbda_region].invoke(
         function_name=fn_arn,
-        qualifier=None,
+        qualifier=None,  # type: ignore[arg-type]
         body=body,
         headers=dict(),
         response_headers=dict(),

moto/iam/models.py

@@ -675,7 +675,7 @@ def __init__(
         self.tags = tags
         # last_used should be treated as part of the public API
         # https://github.com/getmoto/moto/issues/6859
-        self.last_used = None
+        self.last_used: Optional[datetime] = None
         self.last_used_region = None
         self.description = description
         self.permissions_boundary: Optional[str] = permissions_boundary

moto/iot/models.py

@@ -1255,7 +1255,7 @@ def _get_principal(self, principal_arn: str) -> Any:
         cognito = cognitoidentity_backends[self.account_id][self.region_name]
         identities = []
         for identity_pool in cognito.identity_pools:
-            pool_identities = cognito.pools_identities.get(identity_pool, [])
+            pool_identities = cognito.pools_identities.get(identity_pool, {})
             identities.extend(
                 [pi["IdentityId"] for pi in pool_identities.get("Identities", [])]
             )

moto/logs/models.py

@@ -1266,7 +1266,7 @@ def create_export_task(
                 logGroupName, [], fromTime, to, None, None, "", False
             )
             s3_backends[self.account_id]["global"].put_object(
-                destination, destinationPrefix, json.dumps(logs)
+                destination, destinationPrefix, json.dumps(logs).encode("utf-8")
             )
             self.export_tasks[task_id].status["code"] = "completed"
             self.export_tasks[task_id].status["message"] = "Task is completed"

moto/moto_server/werkzeug_app.py

@@ -57,7 +57,9 @@ class DomainDispatcherApplication:
     """
 
     def __init__(
-        self, create_app: Callable[[str], Flask], service: Optional[str] = None
+        self,
+        create_app: Callable[[backends.SERVICE_NAMES], Flask],
+        service: Optional[str] = None,
     ):
         self.create_app = create_app
         self.lock = Lock()
@@ -260,7 +262,7 @@ def __call__(self, environ: Dict[str, Any], start_response: Any) -> Any:
         return backend_app(environ, start_response)
 
 
-def create_backend_app(service: str) -> Flask:
+def create_backend_app(service: backends.SERVICE_NAMES) -> Flask:
     from werkzeug.routing import Map
 
     current_file = os.path.abspath(__file__)
@@ -292,7 +294,7 @@ def create_backend_app(service: str) -> Flask:
     for url_path, handler in backend.flask_paths.items():
         view_func = convert_to_flask_response(handler)
         if handler.__name__ == "dispatch":
-            endpoint = f"{handler.__self__.__name__}.dispatch"
+            endpoint = f"{handler.__self__.__name__}.dispatch"  # type: ignore[attr-defined]
         else:
             endpoint = view_func.__name__