Techdebt: MyPy EC2 (d-e-models) (#5898)

moto/ebs/models.py

@@ -71,7 +71,7 @@ def ec2_backend(self) -> EC2Backend:
         return ec2_backends[self.account_id][self.region_name]
 
     def start_snapshot(
-        self, volume_size: str, tags: Optional[List[Dict[str, str]]], description: str
+        self, volume_size: int, tags: Optional[List[Dict[str, str]]], description: str
     ) -> EBSSnapshot:
         zone_name = f"{self.region_name}a"
         vol = self.ec2_backend.create_volume(size=volume_size, zone_name=zone_name)

moto/ec2/exceptions.py

@@ -1,5 +1,5 @@
 from moto.core.exceptions import RESTError
-from typing import List, Optional, Union
+from typing import Any, List, Optional, Union, Iterable
 
 
 # EC2 has a custom root-tag - <Response> vs <ErrorResponse>
@@ -51,7 +51,7 @@ def __init__(self, parameter):
 
 
 class InvalidDHCPOptionsIdError(EC2ClientError):
-    def __init__(self, dhcp_options_id):
+    def __init__(self, dhcp_options_id: str):
         super().__init__(
             "InvalidDhcpOptionID.NotFound",
             f"DhcpOptionID {dhcp_options_id} does not exist.",
@@ -69,7 +69,7 @@ def __init__(self, msg):
 
 
 class MalformedDHCPOptionsIdError(EC2ClientError):
-    def __init__(self, dhcp_options_id):
+    def __init__(self, dhcp_options_id: Optional[str]):
         super().__init__(
             "InvalidDhcpOptionsId.Malformed",
             f'Invalid id: "{dhcp_options_id}" (expecting "dopt-...")',
@@ -166,15 +166,15 @@ def __init__(self, customer_gateway_id: str):
 
 
 class InvalidNetworkInterfaceIdError(EC2ClientError):
-    def __init__(self, eni_id):
+    def __init__(self, eni_id: str):
         super().__init__(
             "InvalidNetworkInterfaceID.NotFound",
             f"The network interface ID '{eni_id}' does not exist",
         )
 
 
 class InvalidNetworkAttachmentIdError(EC2ClientError):
-    def __init__(self, attachment_id):
+    def __init__(self, attachment_id: str):
         super().__init__(
             "InvalidAttachmentID.NotFound",
             f"The network interface attachment ID '{attachment_id}' does not exist",
@@ -274,7 +274,7 @@ def __init__(self, ami_id: str):
 
 
 class InvalidAMIAttributeItemValueError(EC2ClientError):
-    def __init__(self, attribute: str, value: str):
+    def __init__(self, attribute: str, value: Union[str, Iterable[str], None]):
         super().__init__(
             "InvalidAMIAttributeItemValue",
             f'Invalid attribute item value "{value}" for {attribute} item type.',
@@ -289,52 +289,52 @@ def __init__(self, ami_id: List[str]):
 
 
 class InvalidSnapshotIdError(EC2ClientError):
-    def __init__(self):
+    def __init__(self) -> None:
         # Note: AWS returns empty message for this, as of 2014.08.22.
         super().__init__("InvalidSnapshot.NotFound", "")
 
 
 class InvalidSnapshotInUse(EC2ClientError):
-    def __init__(self, snapshot_id, ami_id):
+    def __init__(self, snapshot_id: str, ami_id: str):
         super().__init__(
             "InvalidSnapshot.InUse",
             f"The snapshot {snapshot_id} is currently in use by {ami_id}",
         )
 
 
 class InvalidVolumeIdError(EC2ClientError):
-    def __init__(self, volume_id):
+    def __init__(self, volume_id: Any):
         super().__init__(
             "InvalidVolume.NotFound", f"The volume '{volume_id}' does not exist."
         )
 
 
 class InvalidVolumeAttachmentError(EC2ClientError):
-    def __init__(self, volume_id, instance_id):
+    def __init__(self, volume_id: str, instance_id: str):
         super().__init__(
             "InvalidAttachment.NotFound",
             f"Volume {volume_id} can not be detached from {instance_id} because it is not attached",
         )
 
 
 class InvalidVolumeDetachmentError(EC2ClientError):
-    def __init__(self, volume_id, instance_id, device):
+    def __init__(self, volume_id: str, instance_id: str, device: str):
         super().__init__(
             "InvalidAttachment.NotFound",
             f"The volume {volume_id} is not attached to instance {instance_id} as device {device}",
         )
 
 
 class VolumeInUseError(EC2ClientError):
-    def __init__(self, volume_id, instance_id):
+    def __init__(self, volume_id: str, instance_id: str):
         super().__init__(
             "VolumeInUse",
             f"Volume {volume_id} is currently attached to {instance_id}",
         )
 
 
 class InvalidAddressError(EC2ClientError):
-    def __init__(self, ip):
+    def __init__(self, ip: Any):
         super().__init__("InvalidAddress.NotFound", f"Address '{ip}' not found.")
 
 
@@ -347,15 +347,15 @@ def __init__(self, bucket_name):
 
 
 class InvalidAllocationIdError(EC2ClientError):
-    def __init__(self, allocation_id):
+    def __init__(self, allocation_id: Any):
         super().__init__(
             "InvalidAllocationID.NotFound",
             f"Allocation ID '{allocation_id}' not found.",
         )
 
 
 class InvalidAssociationIdError(EC2ClientError):
-    def __init__(self, association_id):
+    def __init__(self, association_id: Any):
         super().__init__(
             "InvalidAssociationID.NotFound",
             f"Association ID '{association_id}' not found.",
@@ -426,7 +426,7 @@ def __init__(self, parameter):
 
 
 class InvalidParameterValueError(EC2ClientError):
-    def __init__(self, parameter_value):
+    def __init__(self, parameter_value: str):
         super().__init__(
             "InvalidParameterValue",
             f"Value {parameter_value} is invalid for parameter.",
@@ -480,7 +480,7 @@ def __init__(self, internet_gateway_id, vpc_id):
 
 
 class ResourceAlreadyAssociatedError(EC2ClientError):
-    def __init__(self, resource_id):
+    def __init__(self, resource_id: str):
         super().__init__(
             "Resource.AlreadyAssociated",
             f"Resource {resource_id} is already associated.",
@@ -662,7 +662,7 @@ def __init__(self, name):
 
 
 class InvalidParameterDependency(EC2ClientError):
-    def __init__(self, param, param_needed):
+    def __init__(self, param: str, param_needed: str):
         super().__init__(
             "InvalidParameterDependency",
             f"The parameter [{param}] requires the parameter {param_needed} to be set.",

moto/ec2/models/__init__.py

@@ -186,7 +186,7 @@ def default_vpc_endpoint_service(service_region, zones):
     def raise_error(self, code, message):
         raise EC2ClientError(code, message)
 
-    def raise_not_implemented_error(self, blurb):
+    def raise_not_implemented_error(self, blurb: str):
         raise MotoNotImplementedError(blurb)
 
     def do_resources_exist(self, resource_ids):

moto/ec2/models/dhcp_options.py

@@ -1,4 +1,5 @@
 import itertools
+from typing import Any, Dict, List, Optional
 from ..exceptions import (
     DependencyViolationError,
     InvalidDHCPOptionsIdError,
@@ -12,12 +13,12 @@
 class DHCPOptionsSet(TaggedEC2Resource):
     def __init__(
         self,
-        ec2_backend,
-        domain_name_servers=None,
-        domain_name=None,
-        ntp_servers=None,
-        netbios_name_servers=None,
-        netbios_node_type=None,
+        ec2_backend: Any,
+        domain_name_servers: Optional[List[str]] = None,
+        domain_name: Optional[str] = None,
+        ntp_servers: Optional[List[str]] = None,
+        netbios_name_servers: Optional[List[str]] = None,
+        netbios_node_type: Optional[str] = None,
     ):
         self.ec2_backend = ec2_backend
         self._options = {
@@ -30,7 +31,9 @@ def __init__(
         self.id = random_dhcp_option_id()
         self.vpc = None
 
-    def get_filter_value(self, filter_name):
+    def get_filter_value(
+        self, filter_name: str, method_name: Optional[str] = None
+    ) -> Any:
         """
         API Version 2015-10-01 defines the following filters for DescribeDhcpOptions:
 
@@ -54,26 +57,26 @@ def get_filter_value(self, filter_name):
             return super().get_filter_value(filter_name, "DescribeDhcpOptions")
 
     @property
-    def options(self):
+    def options(self) -> Dict[str, Any]:  # type: ignore[misc]
         return self._options
 
 
 class DHCPOptionsSetBackend:
-    def __init__(self):
-        self.dhcp_options_sets = {}
+    def __init__(self) -> None:
+        self.dhcp_options_sets: Dict[str, DHCPOptionsSet] = {}
 
-    def associate_dhcp_options(self, dhcp_options, vpc):
+    def associate_dhcp_options(self, dhcp_options: DHCPOptionsSet, vpc: Any) -> None:
         dhcp_options.vpc = vpc
         vpc.dhcp_options = dhcp_options
 
     def create_dhcp_options(
         self,
-        domain_name_servers=None,
-        domain_name=None,
-        ntp_servers=None,
-        netbios_name_servers=None,
-        netbios_node_type=None,
-    ):
+        domain_name_servers: Optional[List[str]] = None,
+        domain_name: Optional[str] = None,
+        ntp_servers: Optional[List[str]] = None,
+        netbios_name_servers: Optional[List[str]] = None,
+        netbios_node_type: Optional[str] = None,
+    ) -> DHCPOptionsSet:
 
         NETBIOS_NODE_TYPES = [1, 2, 4, 8]
 
@@ -95,7 +98,7 @@ def create_dhcp_options(
         self.dhcp_options_sets[options.id] = options
         return options
 
-    def delete_dhcp_options_set(self, options_id):
+    def delete_dhcp_options_set(self, options_id: Optional[str]) -> None:
         if not (options_id and options_id.startswith("dopt-")):
             raise MalformedDHCPOptionsIdError(options_id)
 
@@ -105,10 +108,11 @@ def delete_dhcp_options_set(self, options_id):
             self.dhcp_options_sets.pop(options_id)
         else:
             raise InvalidDHCPOptionsIdError(options_id)
-        return True
 
-    def describe_dhcp_options(self, dhcp_options_ids=None, filters=None):
-        dhcp_options_sets = self.dhcp_options_sets.copy().values()
+    def describe_dhcp_options(
+        self, dhcp_options_ids: Optional[List[str]] = None, filters: Any = None
+    ) -> List[DHCPOptionsSet]:
+        dhcp_options_sets = list(self.dhcp_options_sets.copy().values())
 
         if dhcp_options_ids:
             dhcp_options_sets = [