Techdebt: MyPy EC2 (k-l-m-models) (#5905)

moto/ec2/exceptions.py

@@ -43,7 +43,7 @@ class DependencyViolationError(EC2ClientError):
 
 
 class MissingParameterError(EC2ClientError):
-    def __init__(self, parameter):
+    def __init__(self, parameter: str):
         super().__init__(
             "MissingParameter",
             f"The request must contain the parameter {parameter}",
@@ -77,21 +77,21 @@ class MalformedDHCPOptionsIdError(EC2ClientError):
 
 
 class InvalidKeyPairNameError(EC2ClientError):
-    def __init__(self, key):
+    def __init__(self, key: Iterable[str]):
         super().__init__(
             "InvalidKeyPair.NotFound", f"The keypair '{key}' does not exist."
         )
 
 
 class InvalidKeyPairDuplicateError(EC2ClientError):
-    def __init__(self, key):
+    def __init__(self, key: str):
         super().__init__(
             "InvalidKeyPair.Duplicate", f"The keypair '{key}' already exists."
         )
 
 
 class InvalidKeyPairFormatError(EC2ClientError):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__(
             "InvalidKeyPair.Format", "Key is not in valid OpenSSH public key format"
         )
@@ -640,7 +640,7 @@ class OperationNotPermitted4(EC2ClientError):
 
 
 class InvalidLaunchTemplateNameAlreadyExistsError(EC2ClientError):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__(
             "InvalidLaunchTemplateName.AlreadyExistsException",
             "Launch template name already in use.",
@@ -648,7 +648,7 @@ class InvalidLaunchTemplateNameAlreadyExistsError(EC2ClientError):
 
 
 class InvalidLaunchTemplateNameNotFoundError(EC2ClientError):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__(
             "InvalidLaunchTemplateName.NotFoundException",
             "At least one of the launch templates specified in the request does not exist.",
@@ -656,7 +656,7 @@ class InvalidLaunchTemplateNameNotFoundError(EC2ClientError):
 
 
 class InvalidLaunchTemplateNameNotFoundWithNameError(EC2ClientError):
-    def __init__(self, name):
+    def __init__(self, name: str):
         super().__init__(
             "InvalidLaunchTemplateName.NotFoundException",
             f"The specified launch template, with template name {name}, does not exist",

moto/ec2/models/key_pairs.py

@@ -1,3 +1,5 @@
+from typing import Any, Dict, List
+
 from moto.core import BaseModel
 from ..exceptions import (
     FilterNotImplementedError,
@@ -15,13 +17,13 @@ from ..utils import (
 
 
 class KeyPair(BaseModel):
-    def __init__(self, name, fingerprint, material):
+    def __init__(self, name: str, fingerprint: str, material: str):
         self.id = random_key_pair_id()
         self.name = name
         self.fingerprint = fingerprint
         self.material = material
 
-    def get_filter_value(self, filter_name):
+    def get_filter_value(self, filter_name: str) -> str:
         if filter_name == "key-name":
             return self.name
         elif filter_name == "fingerprint":
@@ -31,23 +33,22 @@ class KeyPair(BaseModel):
 
 
 class KeyPairBackend:
-    def __init__(self):
+    def __init__(self) -> None:
-        self.keypairs = {}
+        self.keypairs: Dict[str, KeyPair] = {}
 
-    def create_key_pair(self, name):
+    def create_key_pair(self, name: str) -> KeyPair:
         if name in self.keypairs:
             raise InvalidKeyPairDuplicateError(name)
         keypair = KeyPair(name, **random_key_pair())
         self.keypairs[name] = keypair
         return keypair
 
-    def delete_key_pair(self, name):
+    def delete_key_pair(self, name: str) -> None:
-        if name in self.keypairs:
+        self.keypairs.pop(name, None)
-            self.keypairs.pop(name)
-        return True
 
-    def describe_key_pairs(self, key_names=None, filters=None):
+    def describe_key_pairs(
-        results = []
+        self, key_names: List[str], filters: Any = None
+    ) -> List[KeyPair]:
         if any(key_names):
             results = [
                 keypair
@@ -55,17 +56,17 @@ class KeyPairBackend:
                 if keypair.name in key_names
             ]
             if len(key_names) > len(results):
-                unknown_keys = set(key_names) - set(results)
+                unknown_keys = set(key_names) - set(results)  # type: ignore
                 raise InvalidKeyPairNameError(unknown_keys)
         else:
-            results = self.keypairs.values()
+            results = list(self.keypairs.values())
 
         if filters:
             return generic_filter(filters, results)
         else:
             return results
 
-    def import_key_pair(self, key_name, public_key_material):
+    def import_key_pair(self, key_name: str, public_key_material: str) -> KeyPair:
         if key_name in self.keypairs:
             raise InvalidKeyPairDuplicateError(key_name)
 

moto/ec2/models/launch_templates.py

@@ -1,4 +1,5 @@
 from collections import OrderedDict
+from typing import Any, Dict, List, Optional
 
 from moto.core import CloudFormationModel
 from .core import TaggedEC2Resource
@@ -16,8 +17,14 @@ from ..exceptions import (
 )
 
 
-class LaunchTemplateVersion(object):
+class LaunchTemplateVersion:
-    def __init__(self, template, number, data, description):
+    def __init__(
+        self,
+        template: "LaunchTemplate",
+        number: int,
+        data: Dict[str, Any],
+        description: str,
+    ):
         self.template = template
         self.number = number
         self.data = data
@@ -25,85 +32,101 @@ class LaunchTemplateVersion(object):
         self.create_time = utc_date_and_time()
 
     @property
-    def image_id(self):
+    def image_id(self) -> str:
         return self.data.get("ImageId", "")
 
     @property
-    def instance_type(self):
+    def instance_type(self) -> str:
         return self.data.get("InstanceType", "")
 
     @property
-    def security_groups(self):
+    def security_groups(self) -> List[str]:
         return self.data.get("SecurityGroups", [])
 
     @property
-    def user_data(self):
+    def user_data(self) -> str:
         return self.data.get("UserData", "")
 
 
 class LaunchTemplate(TaggedEC2Resource, CloudFormationModel):
-    def __init__(self, backend, name, template_data, version_description, tag_spec):
+    def __init__(
+        self,
+        backend: Any,
+        name: str,
+        template_data: Dict[str, Any],
+        version_description: str,
+        tag_spec: Dict[str, Dict[str, str]],
+    ):
         self.ec2_backend = backend
         self.name = name
         self.id = random_launch_template_id()
         self.create_time = utc_date_and_time()
-        tag_map = tag_spec.get("launch-template", {})
+        tag_map: Dict[str, str] = tag_spec.get("launch-template", {})
         self.add_tags(tag_map)
         self.tags = self.get_tags()
 
-        self.versions = []
+        self.versions: List[LaunchTemplateVersion] = []
         self.create_version(template_data, version_description)
         self.default_version_number = 1
 
-    def create_version(self, data, description):
+    def create_version(
+        self, data: Dict[str, Any], description: str
+    ) -> LaunchTemplateVersion:
         num = len(self.versions) + 1
         version = LaunchTemplateVersion(self, num, data, description)
         self.versions.append(version)
         return version
 
-    def is_default(self, version):
+    def is_default(self, version: LaunchTemplateVersion) -> bool:
-        return self.default_version == version.number
+        return self.default_version == version.number  # type: ignore
 
-    def get_version(self, num):
+    def get_version(self, num: Any) -> LaunchTemplateVersion:
         if str(num).lower() == "$latest":
             return self.versions[-1]
         if str(num).lower() == "$default":
             return self.default_version()
         return self.versions[int(num) - 1]
 
-    def default_version(self):
+    def default_version(self) -> LaunchTemplateVersion:
         return self.versions[self.default_version_number - 1]
 
-    def latest_version(self):
+    def latest_version(self) -> LaunchTemplateVersion:
         return self.versions[-1]
 
     @property
-    def latest_version_number(self):
+    def latest_version_number(self) -> int:
         return self.latest_version().number
 
     @property
-    def physical_resource_id(self):
+    def physical_resource_id(self) -> str:
         return self.id
 
-    def get_filter_value(self, filter_name):
+    def get_filter_value(
+        self, filter_name: str, method_name: Optional[str] = None
+    ) -> Any:
         if filter_name == "launch-template-name":
             return self.name
         else:
             return super().get_filter_value(filter_name, "DescribeLaunchTemplates")
 
     @staticmethod
-    def cloudformation_name_type():
+    def cloudformation_name_type() -> str:
         return "LaunchTemplateName"
 
     @staticmethod
-    def cloudformation_type():
+    def cloudformation_type() -> str:
         # https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-ec2-launchtemplate.html
         return "AWS::EC2::LaunchTemplate"
 
     @classmethod
-    def create_from_cloudformation_json(
+    def create_from_cloudformation_json(  # type: ignore[misc]
-        cls, resource_name, cloudformation_json, account_id, region_name, **kwargs
+        cls,
-    ):
+        resource_name: str,
+        cloudformation_json: Any,
+        account_id: str,
+        region_name: str,
+        **kwargs: Any
+    ) -> "LaunchTemplate":
 
         from ..models import ec2_backends
 
@@ -124,14 +147,14 @@ class LaunchTemplate(TaggedEC2Resource, CloudFormationModel):
         return launch_template
 
     @classmethod
-    def update_from_cloudformation_json(
+    def update_from_cloudformation_json(  # type: ignore[misc]
         cls,
-        original_resource,
+        original_resource: Any,
-        new_resource_name,
+        new_resource_name: str,
-        cloudformation_json,
+        cloudformation_json: Any,
-        account_id,
+        account_id: str,
-        region_name,
+        region_name: str,
-    ):
+    ) -> "LaunchTemplate":
 
         from ..models import ec2_backends
 
@@ -150,9 +173,13 @@ class LaunchTemplate(TaggedEC2Resource, CloudFormationModel):
         return launch_template
 
     @classmethod
-    def delete_from_cloudformation_json(
+    def delete_from_cloudformation_json(  # type: ignore[misc]
-        cls, resource_name, cloudformation_json, account_id, region_name
+        cls,
-    ):
+        resource_name: str,
+        cloudformation_json: Any,
+        account_id: str,
+        region_name: str,
+    ) -> None:
 
         from ..models import ec2_backends
 
@@ -166,12 +193,18 @@ class LaunchTemplate(TaggedEC2Resource, CloudFormationModel):
 
 
 class LaunchTemplateBackend:
-    def __init__(self):
+    def __init__(self) -> None:
-        self.launch_template_name_to_ids = {}
+        self.launch_template_name_to_ids: Dict[str, str] = {}
-        self.launch_templates = OrderedDict()
+        self.launch_templates: Dict[str, LaunchTemplate] = OrderedDict()
-        self.launch_template_insert_order = []
+        self.launch_template_insert_order: List[str] = []
 
-    def create_launch_template(self, name, description, template_data, tag_spec):
+    def create_launch_template(
+        self,
+        name: str,
+        description: str,
+        template_data: Dict[str, Any],
+        tag_spec: Dict[str, Any],
+    ) -> LaunchTemplate:
         if name in self.launch_template_name_to_ids:
             raise InvalidLaunchTemplateNameAlreadyExistsError()
         template = LaunchTemplate(self, name, template_data, description, tag_spec)
@@ -188,7 +221,7 @@ class LaunchTemplateBackend:
             raise InvalidLaunchTemplateNameNotFoundWithNameError(name)
         return self.get_launch_template(self.launch_template_name_to_ids[name])
 
-    def delete_launch_template(self, name, tid):
+    def delete_launch_template(self, name: str, tid: Optional[str]) -> LaunchTemplate:
         if name:
             tid = self.launch_template_name_to_ids.get(name)
         if tid is None:
@@ -200,8 +233,11 @@ class LaunchTemplateBackend:
         return template
 
     def describe_launch_templates(
-        self, template_names=None, template_ids=None, filters=None
+        self,
-    ):
+        template_names: Optional[List[str]] = None,
+        template_ids: Optional[List[str]] = None,
+        filters: Any = None,
+    ) -> List[LaunchTemplate]:
         if template_names and not template_ids:
             template_ids = []
             for name in template_names:

moto/ec2/models/managed_prefixes.py

@@ -1,3 +1,5 @@
+from typing import Any, Dict, List, Optional
+
 from moto.utilities.utils import filter_resources
 from .core import TaggedEC2Resource
 from ..utils import random_managed_prefix_list_id, describe_tag_filter
@@ -6,14 +8,14 @@ from ..utils import random_managed_prefix_list_id, describe_tag_filter
 class ManagedPrefixList(TaggedEC2Resource):
     def __init__(
         self,
-        backend,
+        backend: Any,
-        address_family=None,
+        region: str,
-        entry=None,
+        address_family: Optional[str] = None,
-        max_entries=None,
+        entry: Optional[List[Dict[str, str]]] = None,
-        prefix_list_name=None,
+        max_entries: Optional[str] = None,
-        region=None,
+        prefix_list_name: Optional[str] = None,
-        tags=None,
+        tags: Optional[Dict[str, str]] = None,
-        owner_id=None,
+        owner_id: Optional[str] = None,
     ):
         self.ec2_backend = backend
         self.address_family = address_family
@@ -23,49 +25,51 @@ class ManagedPrefixList(TaggedEC2Resource):
         self.state = "create-complete"
         self.state_message = "create complete"
         self.add_tags(tags or {})
-        self.version = 1
+        self.version: Optional[int] = 1
         self.entries = {self.version: entry} if entry else {}
         self.resource_owner_id = owner_id if owner_id else None
         self.prefix_list_arn = self.arn(region, self.owner_id)
         self.delete_counter = 1
 
-    def arn(self, region, owner_id):
+    def arn(self, region: str, owner_id: str) -> str:
         return f"arn:aws:ec2:{region}:{owner_id}:prefix-list/{self.id}"
 
     @property
-    def owner_id(self):
+    def owner_id(self) -> str:
         return self.resource_owner_id or self.ec2_backend.account_id
 
 
 class ManagedPrefixListBackend:
-    def __init__(self):
+    def __init__(self) -> None:
-        self.managed_prefix_lists = {}
+        self.managed_prefix_lists: Dict[str, ManagedPrefixList] = {}
         self.create_default_pls()
 
     def create_managed_prefix_list(
         self,
-        address_family=None,
+        address_family: Optional[str] = None,
-        entry=None,
+        entry: Optional[List[Dict[str, str]]] = None,
-        max_entries=None,
+        max_entries: Optional[str] = None,
-        prefix_list_name=None,
+        prefix_list_name: Optional[str] = None,
-        tags=None,
+        tags: Optional[Dict[str, str]] = None,
-        owner_id=None,
+        owner_id: Optional[str] = None,
-    ):
+    ) -> ManagedPrefixList:
         managed_prefix_list = ManagedPrefixList(
             self,
             address_family=address_family,
             entry=entry,
             max_entries=max_entries,
             prefix_list_name=prefix_list_name,
-            region=self.region_name,
+            region=self.region_name,  # type: ignore[attr-defined]
             tags=tags,
             owner_id=owner_id,
         )
         self.managed_prefix_lists[managed_prefix_list.id] = managed_prefix_list
         return managed_prefix_list
 
-    def describe_managed_prefix_lists(self, prefix_list_ids=None, filters=None):
+    def describe_managed_prefix_lists(
-        managed_prefix_lists = list(self.managed_prefix_lists.copy().values())
+        self, prefix_list_ids: Optional[List[str]] = None, filters: Any = None
+    ) -> List[ManagedPrefixList]:
+        managed_prefix_lists = list(self.managed_prefix_lists.values())
         attr_pairs = (
             ("owner-id", "owner_id"),
             ("prefix-list-id", "id"),
@@ -92,29 +96,30 @@ class ManagedPrefixListBackend:
                 item.delete_counter -= 1
         return result
 
-    def get_managed_prefix_list_entries(self, prefix_list_id=None):
+    def get_managed_prefix_list_entries(
-        managed_prefix_list = self.managed_prefix_lists.get(prefix_list_id)
+        self, prefix_list_id: str
-        return managed_prefix_list
+    ) -> Optional[ManagedPrefixList]:
+        return self.managed_prefix_lists.get(prefix_list_id)
 
-    def delete_managed_prefix_list(self, prefix_list_id):
+    def delete_managed_prefix_list(self, prefix_list_id: str) -> ManagedPrefixList:
-        managed_prefix_list = self.managed_prefix_lists.get(prefix_list_id)
+        managed_prefix_list: ManagedPrefixList = self.managed_prefix_lists.get(prefix_list_id)  # type: ignore
         managed_prefix_list.state = "delete-complete"
         return managed_prefix_list
 
     def modify_managed_prefix_list(
         self,
-        add_entry=None,
+        add_entry: List[Dict[str, str]],
-        prefix_list_id=None,
+        remove_entry: List[Dict[str, str]],
-        current_version=None,
+        prefix_list_id: Optional[str] = None,
-        prefix_list_name=None,
+        current_version: Optional[str] = None,
-        remove_entry=None,
+        prefix_list_name: Optional[str] = None,
-    ):
+    ) -> ManagedPrefixList:
-        managed_pl = self.managed_prefix_lists.get(prefix_list_id)
+        managed_pl: ManagedPrefixList = self.managed_prefix_lists.get(prefix_list_id)  # type: ignore
         managed_pl.prefix_list_name = prefix_list_name
         if remove_entry or add_entry:
-            latest_version = managed_pl.entries.get(managed_pl.version)
+            latest_version = managed_pl.entries.get(managed_pl.version)  # type: ignore[arg-type]
             entries = (
-                managed_pl.entries.get(current_version, latest_version).copy()
+                managed_pl.entries.get(current_version, latest_version).copy()  # type: ignore
                 if managed_pl.entries
                 else []
             )
@@ -125,12 +130,12 @@ class ManagedPrefixListBackend:
             for item in add_entry:
                 if item not in entries.copy():
                     entries.append(item)
-            managed_pl.version += 1
+            managed_pl.version += 1  # type: ignore[operator]
             managed_pl.entries[managed_pl.version] = entries
         managed_pl.state = "modify-complete"
         return managed_pl
 
-    def create_default_pls(self):
+    def create_default_pls(self) -> None:
         entry = [
             {"Cidr": "52.216.0.0/15", "Description": "default"},
             {"Cidr": "3.5.0.0/19", "Description": "default"},
@@ -140,7 +145,7 @@ class ManagedPrefixListBackend:
         managed_prefix_list = self.create_managed_prefix_list(
             address_family="IPv4",
             entry=entry,
-            prefix_list_name=f"com.amazonaws.{self.region_name}.s3",
+            prefix_list_name=f"com.amazonaws.{self.region_name}.s3",  # type: ignore[attr-defined]
             owner_id="aws",
         )
         managed_prefix_list.version = None
@@ -157,7 +162,7 @@ class ManagedPrefixListBackend:
         managed_prefix_list = self.create_managed_prefix_list(
             address_family="IPv4",
             entry=entry,
-            prefix_list_name=f"com.amazonaws.{self.region_name}.dynamodb",
+            prefix_list_name=f"com.amazonaws.{self.region_name}.dynamodb",  # type: ignore[attr-defined]
             owner_id="aws",
         )
         managed_prefix_list.version = None

moto/ec2/responses/key_pairs.py

@@ -12,10 +12,8 @@ class KeyPairs(EC2BaseResponse):
     def delete_key_pair(self):
         name = self._get_param("KeyName")
         if self.is_not_dryrun("DeleteKeyPair"):
-            success = str(self.ec2_backend.delete_key_pair(name)).lower()
+            self.ec2_backend.delete_key_pair(name)
-            return self.response_template(DELETE_KEY_PAIR_RESPONSE).render(
+            return self.response_template(DELETE_KEY_PAIR_RESPONSE).render()
-                success=success
-            )
 
     def describe_key_pairs(self):
         names = self._get_multi_param("KeyName")
@@ -57,7 +55,7 @@ CREATE_KEY_PAIR_RESPONSE = """<CreateKeyPairResponse xmlns="http://ec2.amazonaws
 
 DELETE_KEY_PAIR_RESPONSE = """<DeleteKeyPairResponse xmlns="http://ec2.amazonaws.com/doc/2013-10-15/">
   <requestId>59dbff89-35bd-4eac-99ed-be587EXAMPLE</requestId>
-  <return>{{ success }}</return>
+  <return>true</return>
 </DeleteKeyPairResponse>"""
 
 IMPORT_KEYPAIR_RESPONSE = """<?xml version="1.0" encoding="UTF-8"?>

moto/ec2/utils.py

@@ -150,7 +150,7 @@ def random_volume_id() -> str:
     return random_id(prefix=EC2_RESOURCE_TO_PREFIX["volume"])
 
 
-def random_key_pair_id():
+def random_key_pair_id() -> str:
     return random_id(prefix=EC2_RESOURCE_TO_PREFIX["key-pair"])
 
 
@@ -224,7 +224,7 @@ def random_transit_gateway_attachment_id():
     )
 
 
-def random_launch_template_id():
+def random_launch_template_id() -> str:
     return random_id(prefix=EC2_RESOURCE_TO_PREFIX["launch-template"], size=17)
 
 
@@ -290,7 +290,7 @@ def generate_route_id(
     return f"{route_table_id}~{cidr_block}"
 
 
-def random_managed_prefix_list_id():
+def random_managed_prefix_list_id() -> str:
     return random_id(prefix=EC2_RESOURCE_TO_PREFIX["managed-prefix-list"], size=8)
 
 
@@ -547,7 +547,7 @@ def simple_aws_filter_to_re(filter_string):
     return tmp_filter
 
 
-def random_key_pair():
+def random_key_pair() -> Dict[str, str]:
     private_key = rsa.generate_private_key(
         public_exponent=65537, key_size=2048, backend=default_backend()
     )
@@ -639,7 +639,7 @@ def generate_instance_identity_document(instance):
     return document
 
 
-def rsa_public_key_parse(key_material):
+def rsa_public_key_parse(key_material: Any) -> Any:
     # These imports take ~.5s; let's keep them local
     import sshpubkeys.exceptions
     from sshpubkeys.keys import SSHKey
@@ -659,7 +659,7 @@ def rsa_public_key_parse(key_material):
     return public_key.rsa
 
 
-def rsa_public_key_fingerprint(rsa_public_key):
+def rsa_public_key_fingerprint(rsa_public_key: Any) -> str:
     key_data = rsa_public_key.public_bytes(
         encoding=serialization.Encoding.DER,
         format=serialization.PublicFormat.SubjectPublicKeyInfo,
@@ -719,7 +719,9 @@ def filter_iam_instance_profiles(
     return instance_profile
 
 
-def describe_tag_filter(filters, instances):
+def describe_tag_filter(
+    filters: Any, instances: List[FILTER_TYPE]
+) -> List[FILTER_TYPE]:
     result = instances.copy()
     for instance in instances:
         for key in filters:

setup.cfg

@@ -229,7 +229,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*,moto/c*,moto/d*,moto/ebs/,moto/ec2/models/a*,moto/ec2/models/c*,moto/ec2/models/d*,moto/ec2/models/e*,moto/ec2/models/f*,moto/ec2/models/h*,moto/ec2/models/i*,moto/moto_api
+files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2/models/a*,moto/ec2/models/c*,moto/ec2/models/d*,moto/ec2/models/e*,moto/ec2/models/f*,moto/ec2/models/h*,moto/ec2/models/i*,moto/ec2/models/k*,moto/ec2/models/l*,moto/ec2/models/m*,moto/moto_api
 show_column_numbers=True
 show_error_codes = True
 disable_error_code=abstract