Techdebt: MyPy ES (#5938)

moto/ec2instanceconnect/models.py

@@ -3,7 +3,7 @@
 
 
 class Ec2InstanceConnectBackend(BaseBackend):
-    def send_ssh_public_key(self):
+    def send_ssh_public_key(self) -> str:
         return json.dumps(
             {"RequestId": "example-2a47-4c91-9700-e37e85162cb6", "Success": True}
         )

moto/ec2instanceconnect/responses.py

@@ -1,14 +1,14 @@
 from moto.core.responses import BaseResponse
-from .models import ec2instanceconnect_backends
+from .models import ec2instanceconnect_backends, Ec2InstanceConnectBackend
 
 
 class Ec2InstanceConnectResponse(BaseResponse):
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__(service_name="ec2-instanceconnect")
 
     @property
-    def ec2instanceconnect_backend(self):
+    def ec2instanceconnect_backend(self) -> Ec2InstanceConnectBackend:
         return ec2instanceconnect_backends[self.current_account][self.region]
 
-    def send_ssh_public_key(self):
+    def send_ssh_public_key(self) -> str:
         return self.ec2instanceconnect_backend.send_ssh_public_key()

moto/es/exceptions.py

@@ -9,17 +9,17 @@ class ElasticSearchError(JsonRESTError):
 class ResourceNotFound(ElasticSearchError):
     code = 409
 
-    def __init__(self, resource_type, resource_name):
+    def __init__(self, resource_type: str, resource_name: str):
         msg = f"{resource_type} not found: {resource_name}"
         super().__init__("ResourceNotFoundException", msg)
 
 
 class InvalidDomainName(ElasticSearchError):
-    def __init__(self, domain_name):
+    def __init__(self, domain_name: str):
         msg = f"1 validation error detected: Value '{domain_name}' at 'domainName' failed to satisfy constraint: Member must satisfy regular expression pattern: [a-z][a-z0-9\\-]+"
         super().__init__("ValidationException", msg)
 
 
 class DomainNotFound(ResourceNotFound):
-    def __init__(self, domain_name):
+    def __init__(self, domain_name: str):
         super().__init__("Domain", domain_name)

moto/es/models.py

@@ -1,3 +1,4 @@
+from typing import Any, Dict, List
 from moto.core import BaseBackend, BackendDict, BaseModel
 from moto.moto_api._internal import mock_random
 from .exceptions import DomainNotFound
@@ -6,22 +7,22 @@
 class Domain(BaseModel):
     def __init__(
         self,
-        region_name,
-        domain_name,
-        es_version,
-        elasticsearch_cluster_config,
-        ebs_options,
-        access_policies,
-        snapshot_options,
-        vpc_options,
-        cognito_options,
-        encryption_at_rest_options,
-        node_to_node_encryption_options,
-        advanced_options,
-        log_publishing_options,
-        domain_endpoint_options,
-        advanced_security_options,
-        auto_tune_options,
+        region_name: str,
+        domain_name: str,
+        es_version: str,
+        elasticsearch_cluster_config: Dict[str, Any],
+        ebs_options: Dict[str, Any],
+        access_policies: Dict[str, Any],
+        snapshot_options: Dict[str, Any],
+        vpc_options: Dict[str, Any],
+        cognito_options: Dict[str, Any],
+        encryption_at_rest_options: Dict[str, Any],
+        node_to_node_encryption_options: Dict[str, Any],
+        advanced_options: Dict[str, Any],
+        log_publishing_options: Dict[str, Any],
+        domain_endpoint_options: Dict[str, Any],
+        advanced_security_options: Dict[str, Any],
+        auto_tune_options: Dict[str, Any],
     ):
         self.domain_id = mock_random.get_random_hex(8)
         self.region_name = region_name
@@ -44,10 +45,10 @@ def __init__(
             self.auto_tune_options["State"] = "ENABLED"
 
     @property
-    def arn(self):
+    def arn(self) -> str:
         return f"arn:aws:es:{self.region_name}:domain/{self.domain_id}"
 
-    def to_json(self):
+    def to_json(self) -> Dict[str, Any]:
         return {
             "DomainId": self.domain_id,
             "DomainName": self.domain_name,
@@ -76,28 +77,28 @@ def to_json(self):
 class ElasticsearchServiceBackend(BaseBackend):
     """Implementation of ElasticsearchService APIs."""
 
-    def __init__(self, region_name, account_id):
+    def __init__(self, region_name: str, account_id: str):
         super().__init__(region_name, account_id)
-        self.domains = dict()
+        self.domains: Dict[str, Domain] = dict()
 
     def create_elasticsearch_domain(
         self,
-        domain_name,
-        elasticsearch_version,
-        elasticsearch_cluster_config,
-        ebs_options,
-        access_policies,
-        snapshot_options,
-        vpc_options,
-        cognito_options,
-        encryption_at_rest_options,
-        node_to_node_encryption_options,
-        advanced_options,
-        log_publishing_options,
-        domain_endpoint_options,
-        advanced_security_options,
-        auto_tune_options,
-    ):
+        domain_name: str,
+        elasticsearch_version: str,
+        elasticsearch_cluster_config: Dict[str, Any],
+        ebs_options: Dict[str, Any],
+        access_policies: Dict[str, Any],
+        snapshot_options: Dict[str, Any],
+        vpc_options: Dict[str, Any],
+        cognito_options: Dict[str, Any],
+        encryption_at_rest_options: Dict[str, Any],
+        node_to_node_encryption_options: Dict[str, Any],
+        advanced_options: Dict[str, Any],
+        log_publishing_options: Dict[str, Any],
+        domain_endpoint_options: Dict[str, Any],
+        advanced_security_options: Dict[str, Any],
+        auto_tune_options: Dict[str, Any],
+    ) -> Dict[str, Any]:
         # TODO: Persist/Return other attributes
         new_domain = Domain(
             region_name=self.region_name,
@@ -120,17 +121,17 @@ def create_elasticsearch_domain(
         self.domains[domain_name] = new_domain
         return new_domain.to_json()
 
-    def delete_elasticsearch_domain(self, domain_name):
+    def delete_elasticsearch_domain(self, domain_name: str) -> None:
         if domain_name not in self.domains:
             raise DomainNotFound(domain_name)
         del self.domains[domain_name]
 
-    def describe_elasticsearch_domain(self, domain_name):
+    def describe_elasticsearch_domain(self, domain_name: str) -> Dict[str, Any]:
         if domain_name not in self.domains:
             raise DomainNotFound(domain_name)
         return self.domains[domain_name].to_json()
 
-    def list_domain_names(self):
+    def list_domain_names(self) -> List[Dict[str, str]]:
         """
         The engine-type parameter is not yet supported.
         Pagination is not yet implemented.

moto/es/responses.py

@@ -1,46 +1,48 @@
 import json
 import re
+from typing import Any
 
+from moto.core.common_types import TYPE_RESPONSE
 from moto.core.responses import BaseResponse
 from .exceptions import InvalidDomainName
-from .models import es_backends
+from .models import es_backends, ElasticsearchServiceBackend
 
 
 class ElasticsearchServiceResponse(BaseResponse):
     """Handler for ElasticsearchService requests and responses."""
 
-    def __init__(self):
+    def __init__(self) -> None:
         super().__init__(service_name="elasticsearch")
 
     @property
-    def es_backend(self):
+    def es_backend(self) -> ElasticsearchServiceBackend:
         """Return backend instance specific for this region."""
         return es_backends[self.current_account][self.region]
 
     @classmethod
-    def list_domains(cls, request, full_url, headers):
+    def list_domains(cls, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE:  # type: ignore
         response = ElasticsearchServiceResponse()
         response.setup_class(request, full_url, headers)
         if request.method == "GET":
             return response.list_domain_names()
 
     @classmethod
-    def domains(cls, request, full_url, headers):
+    def domains(cls, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE:  # type: ignore
         response = ElasticsearchServiceResponse()
         response.setup_class(request, full_url, headers)
         if request.method == "POST":
             return response.create_elasticsearch_domain()
 
     @classmethod
-    def domain(cls, request, full_url, headers):
+    def domain(cls, request: Any, full_url: str, headers: Any) -> TYPE_RESPONSE:  # type: ignore
         response = ElasticsearchServiceResponse()
         response.setup_class(request, full_url, headers)
         if request.method == "DELETE":
             return response.delete_elasticsearch_domain()
         if request.method == "GET":
             return response.describe_elasticsearch_domain()
 
-    def create_elasticsearch_domain(self):
+    def create_elasticsearch_domain(self) -> TYPE_RESPONSE:
         params = json.loads(self.body)
         domain_name = params.get("DomainName")
         if not re.match(r"^[a-z][a-z0-9\-]+$", domain_name):
@@ -78,12 +80,12 @@ def create_elasticsearch_domain(self):
         )
         return 200, {}, json.dumps({"DomainStatus": domain_status})
 
-    def delete_elasticsearch_domain(self):
+    def delete_elasticsearch_domain(self) -> TYPE_RESPONSE:
         domain_name = self.path.split("/")[-1]
         self.es_backend.delete_elasticsearch_domain(domain_name=domain_name)
         return 200, {}, json.dumps(dict())
 
-    def describe_elasticsearch_domain(self):
+    def describe_elasticsearch_domain(self) -> TYPE_RESPONSE:
         domain_name = self.path.split("/")[-1]
         if not re.match(r"^[a-z][a-z0-9\-]+$", domain_name):
             raise InvalidDomainName(domain_name)
@@ -92,6 +94,6 @@ def describe_elasticsearch_domain(self):
         )
         return 200, {}, json.dumps({"DomainStatus": domain_status})
 
-    def list_domain_names(self):
+    def list_domain_names(self) -> TYPE_RESPONSE:
         domain_names = self.es_backend.list_domain_names()
         return 200, {}, json.dumps({"DomainNames": domain_names})

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/,moto/moto_api
+files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2,moto/ec2instanceconnect,moto/es,moto/moto_api
 show_column_numbers=True
 show_error_codes = True
 disable_error_code=abstract