Techdebt: MyPy EC2 (a-models) (#5888)

moto/ec2/exceptions.py

@@ -1,4 +1,5 @@
 from moto.core.exceptions import RESTError
+from typing import List, Union
 
 
 # EC2 has a custom root-tag - <Response> vs <ErrorResponse>
@@ -258,31 +259,31 @@ def __init__(self, instance_type, error_type="InvalidInstanceType.NotFound"):
 
 
 class InvalidAMIIdError(EC2ClientError):
-    def __init__(self, ami_id):
+    def __init__(self, ami_id: Union[List[str], str]):
         super().__init__(
             "InvalidAMIID.NotFound",
             f"The image id '[{ami_id}]' does not exist",
         )
 
 
 class UnvailableAMIIdError(EC2ClientError):
-    def __init__(self, ami_id):
+    def __init__(self, ami_id: str):
         super().__init__(
             "InvalidAMIID.Unavailable",
             f"The image id '[{ami_id}]' is no longer available",
         )
 
 
 class InvalidAMIAttributeItemValueError(EC2ClientError):
-    def __init__(self, attribute, value):
+    def __init__(self, attribute: str, value: str):
         super().__init__(
             "InvalidAMIAttributeItemValue",
             f'Invalid attribute item value "{value}" for {attribute} item type.',
         )
 
 
 class MalformedAMIIdError(EC2ClientError):
-    def __init__(self, ami_id):
+    def __init__(self, ami_id: List[str]):
         super().__init__(
             "InvalidAMIID.Malformed", f'Invalid id: "{ami_id}" (expecting "ami-...")'
         )
@@ -694,7 +695,7 @@ def __init__(self, vpc_end_point_id):
 
 
 class InvalidTaggableResourceType(EC2ClientError):
-    def __init__(self, resource_type):
+    def __init__(self, resource_type: str):
         super().__init__(
             "InvalidParameterValue",
             f"'{resource_type}' is not a valid taggable resource type for this operation.",

moto/ec2/models/amis.py

@@ -1,6 +1,7 @@
 import json
 import re
 from os import environ
+from typing import Any, Dict, List, Iterable, Optional, Set
 from moto.utilities.utils import load_resource
 from ..exceptions import (
     InvalidAMIIdError,
@@ -10,6 +11,7 @@
     UnvailableAMIIdError,
 )
 from .core import TaggedEC2Resource
+from .instances import Instance
 from ..utils import (
     random_ami_id,
     generic_filter,
@@ -18,7 +20,7 @@
 
 
 if "MOTO_AMIS_PATH" in environ:
-    with open(environ.get("MOTO_AMIS_PATH"), "r", encoding="utf-8") as f:
+    with open(environ["MOTO_AMIS_PATH"], "r", encoding="utf-8") as f:
         AMIS = json.load(f)
 else:
     AMIS = load_resource(__name__, "../resources/amis.json")
@@ -27,28 +29,28 @@
 class Ami(TaggedEC2Resource):
     def __init__(
         self,
-        ec2_backend,
-        ami_id,
-        instance=None,
-        source_ami=None,
-        name=None,
-        description=None,
-        owner_id=None,
-        owner_alias=None,
-        public=False,
-        virtualization_type=None,
-        architecture=None,
-        state="available",
-        creation_date=None,
-        platform=None,
-        image_type="machine",
-        image_location=None,
-        hypervisor=None,
-        root_device_type="standard",
-        root_device_name="/dev/sda1",
-        sriov="simple",
-        region_name="us-east-1a",
-        snapshot_description=None,
+        ec2_backend: Any,
+        ami_id: str,
+        instance: Optional[Instance] = None,
+        source_ami: Optional["Ami"] = None,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+        owner_id: Optional[str] = None,
+        owner_alias: Optional[str] = None,
+        public: bool = False,
+        virtualization_type: Optional[str] = None,
+        architecture: Optional[str] = None,
+        state: str = "available",
+        creation_date: Optional[str] = None,
+        platform: Optional[str] = None,
+        image_type: str = "machine",
+        image_location: Optional[str] = None,
+        hypervisor: Optional[str] = None,
+        root_device_type: str = "standard",
+        root_device_name: str = "/dev/sda1",
+        sriov: str = "simple",
+        region_name: str = "us-east-1a",
+        snapshot_description: Optional[str] = None,
     ):
         self.ec2_backend = ec2_backend
         self.id = ami_id
@@ -92,8 +94,8 @@ def __init__(
             if not description:
                 self.description = source_ami.description
 
-        self.launch_permission_groups = set()
-        self.launch_permission_users = set()
+        self.launch_permission_groups: Set[str] = set()
+        self.launch_permission_users: Set[str] = set()
 
         if public:
             self.launch_permission_groups.add("all")
@@ -109,14 +111,16 @@ def __init__(
         self.ec2_backend.delete_volume(volume.id)
 
     @property
-    def is_public(self):
+    def is_public(self) -> bool:
         return "all" in self.launch_permission_groups
 
     @property
-    def is_public_string(self):
+    def is_public_string(self) -> str:
         return str(self.is_public).lower()
 
-    def get_filter_value(self, filter_name):
+    def get_filter_value(
+        self, filter_name: str, method_name: Optional[str] = None
+    ) -> Any:
         if filter_name == "virtualization-type":
             return self.virtualization_type
         elif filter_name == "kernel-id":
@@ -142,12 +146,12 @@ def get_filter_value(self, filter_name):
 class AmiBackend:
     AMI_REGEX = re.compile("ami-[a-z0-9]+")
 
-    def __init__(self):
-        self.amis = {}
-        self.deleted_amis = list()
+    def __init__(self) -> None:
+        self.amis: Dict[str, Ami] = {}
+        self.deleted_amis: List[str] = list()
         self._load_amis()
 
-    def _load_amis(self):
+    def _load_amis(self) -> None:
         for ami in AMIS:
             ami_id = ami["ami_id"]
             # we are assuming the default loaded amis are owned by amazon
@@ -157,7 +161,7 @@ def _load_amis(self):
         if "MOTO_AMIS_PATH" not in environ:
             try:
                 latest_amis = load_resource(
-                    __name__, f"../resources/latest_amis/{self.region_name}.json"
+                    __name__, f"../resources/latest_amis/{self.region_name}.json"  # type: ignore[attr-defined]
                 )
                 for ami in latest_amis:
                     ami_id = ami["ami_id"]
@@ -169,14 +173,14 @@ def _load_amis(self):
 
     def create_image(
         self,
-        instance_id,
-        name=None,
-        description=None,
-        tag_specifications=None,
-    ):
+        instance_id: str,
+        name: str,
+        description: str,
+        tag_specifications: List[Dict[str, Any]],
+    ) -> Ami:
         # TODO: check that instance exists and pull info from it.
         ami_id = random_ami_id()
-        instance = self.get_instance(instance_id)
+        instance = self.get_instance(instance_id)  # type: ignore[attr-defined]
         tags = []
         for tag_specification in tag_specifications:
             resource_type = tag_specification["ResourceType"]
@@ -202,12 +206,17 @@ def create_image(
         self.amis[ami_id] = ami
         return ami
 
-    def copy_image(self, source_image_id, source_region, name=None, description=None):
+    def copy_image(
+        self,
+        source_image_id: str,
+        source_region: str,
+        name: Optional[str] = None,
+        description: Optional[str] = None,
+    ) -> Ami:
         from ..models import ec2_backends
 
-        source_ami = ec2_backends[self.account_id][source_region].describe_images(
-            ami_ids=[source_image_id]
-        )[0]
+        source_backend = ec2_backends[self.account_id][source_region]  # type: ignore[attr-defined]
+        source_ami = source_backend.describe_images(ami_ids=[source_image_id])[0]
         ami_id = random_ami_id()
         ami = Ami(
             self,
@@ -220,10 +229,16 @@ def copy_image(self, source_image_id, source_region, name=None, description=None
         self.amis[ami_id] = ami
         return ami
 
-    def describe_images(self, ami_ids=(), filters=None, exec_users=None, owners=None):
-        images = self.amis.copy().values()
-
-        if len(ami_ids):
+    def describe_images(
+        self,
+        ami_ids: Optional[List[str]] = None,
+        filters: Optional[Dict[str, Any]] = None,
+        exec_users: Optional[List[str]] = None,
+        owners: Optional[List[str]] = None,
+    ) -> List[Ami]:
+        images = list(self.amis.copy().values())
+
+        if ami_ids and len(ami_ids):
             # boto3 seems to default to just searching based on ami ids if that parameter is passed
             # and if no images are found, it raises an errors
             # Note that we can search for images that have been previously deleted, without raising any errors
@@ -254,7 +269,7 @@ def describe_images(self, ami_ids=(), filters=None, exec_users=None, owners=None
                 # support filtering by Owners=['self']
                 if "self" in owners:
                     owners = list(
-                        map(lambda o: self.account_id if o == "self" else o, owners)
+                        map(lambda o: self.account_id if o == "self" else o, owners)  # type: ignore[attr-defined]
                     )
                 images = [
                     ami
@@ -268,24 +283,26 @@ def describe_images(self, ami_ids=(), filters=None, exec_users=None, owners=None
 
         return images
 
-    def deregister_image(self, ami_id):
+    def deregister_image(self, ami_id: str) -> None:
         if ami_id in self.amis:
             self.amis.pop(ami_id)
             self.deleted_amis.append(ami_id)
-            return True
         elif ami_id in self.deleted_amis:
             raise UnvailableAMIIdError(ami_id)
-        raise InvalidAMIIdError(ami_id)
+        else:
+            raise InvalidAMIIdError(ami_id)
 
-    def get_launch_permission_groups(self, ami_id):
+    def get_launch_permission_groups(self, ami_id: str) -> Iterable[str]:
         ami = self.describe_images(ami_ids=[ami_id])[0]
         return ami.launch_permission_groups
 
-    def get_launch_permission_users(self, ami_id):
+    def get_launch_permission_users(self, ami_id: str) -> Iterable[str]:
         ami = self.describe_images(ami_ids=[ami_id])[0]
         return ami.launch_permission_users
 
-    def validate_permission_targets(self, user_ids=None, group=None):
+    def validate_permission_targets(
+        self, user_ids: Optional[List[str]] = None, group: Optional[str] = None
+    ) -> None:
         # If anything is invalid, nothing is added. (No partial success.)
         if user_ids:
             """
@@ -300,7 +317,12 @@ def validate_permission_targets(self, user_ids=None, group=None):
         if group and group != "all":
             raise InvalidAMIAttributeItemValueError("UserGroup", group)
 
-    def add_launch_permission(self, ami_id, user_ids=None, group=None):
+    def add_launch_permission(
+        self,
+        ami_id: str,
+        user_ids: Optional[List[str]] = None,
+        group: Optional[str] = None,
+    ) -> None:
         ami = self.describe_images(ami_ids=[ami_id])[0]
         self.validate_permission_targets(user_ids=user_ids, group=group)
 
@@ -311,9 +333,9 @@ def add_launch_permission(self, ami_id, user_ids=None, group=None):
         if group:
             ami.launch_permission_groups.add(group)
 
-        return True
-
-    def register_image(self, name=None, description=None):
+    def register_image(
+        self, name: Optional[str] = None, description: Optional[str] = None
+    ) -> Ami:
         ami_id = random_ami_id()
         ami = Ami(
             self,
@@ -326,7 +348,12 @@ def register_image(self, name=None, description=None):
         self.amis[ami_id] = ami
         return ami
 
-    def remove_launch_permission(self, ami_id, user_ids=None, group=None):
+    def remove_launch_permission(
+        self,
+        ami_id: str,
+        user_ids: Optional[List[str]] = None,
+        group: Optional[str] = None,
+    ) -> None:
         ami = self.describe_images(ami_ids=[ami_id])[0]
         self.validate_permission_targets(user_ids=user_ids, group=group)
 
@@ -336,5 +363,3 @@ def remove_launch_permission(self, ami_id, user_ids=None, group=None):
 
         if group:
             ami.launch_permission_groups.discard(group)
-
-        return True

moto/ec2/models/availability_zones_and_regions.py

@@ -1,16 +1,23 @@
 from boto3 import Session
+from typing import Any, Dict, List, Optional
 from moto.utilities.utils import filter_resources
 
 
-class Region(object):
-    def __init__(self, name, endpoint, opt_in_status):
+class Region:
+    def __init__(self, name: str, endpoint: str, opt_in_status: str):
         self.name = name
         self.endpoint = endpoint
         self.opt_in_status = opt_in_status
 
 
-class Zone(object):
-    def __init__(self, name, region_name, zone_id, zone_type="availability-zone"):
+class Zone:
+    def __init__(
+        self,
+        name: str,
+        region_name: str,
+        zone_id: str,
+        zone_type: str = "availability-zone",
+    ):
         self.name = name
         self.region_name = region_name
         self.zone_id = zone_id
@@ -292,7 +299,9 @@ class RegionsAndZonesBackend:
         ],
     }
 
-    def describe_regions(self, region_names=None):
+    def describe_regions(
+        self, region_names: Optional[List[str]] = None
+    ) -> List[Region]:
         if not region_names:
             return self.regions
         ret = []
@@ -302,9 +311,11 @@ def describe_regions(self, region_names=None):
                     ret.append(region)
         return ret
 
-    def describe_availability_zones(self, filters=None):
+    def describe_availability_zones(
+        self, filters: Optional[List[Dict[str, Any]]] = None
+    ) -> List[Zone]:
         # We might not have any zones for the current region, if it was introduced recently
-        zones = self.zones.get(self.region_name, [])
+        zones = self.zones.get(self.region_name, [])  # type: ignore[attr-defined]
         attr_pairs = (
             ("zone-id", "zone_id"),
             ("zone-type", "zone_type"),
@@ -316,7 +327,8 @@ def describe_availability_zones(self, filters=None):
             result = filter_resources(zones, filters, attr_pairs)
         return result
 
-    def get_zone_by_name(self, name):
+    def get_zone_by_name(self, name: str) -> Optional[Zone]:
         for zone in self.describe_availability_zones():
             if zone.name == name:
                 return zone
+        return None