From c2afe19ff17ae0e37b9038ac160534f34716ff0f Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Fri, 17 Feb 2023 10:36:30 -0100 Subject: [PATCH] Techdebt: MyPy ES (#5938) --- moto/ec2instanceconnect/models.py | 2 +- moto/ec2instanceconnect/responses.py | 8 +-- moto/es/exceptions.py | 6 +-- moto/es/models.py | 79 ++++++++++++++-------------- moto/es/responses.py | 22 ++++---- setup.cfg | 2 +- 6 files changed, 61 insertions(+), 58 deletions(-) diff --git a/moto/ec2instanceconnect/models.py b/moto/ec2instanceconnect/models.py index b79babaff..fe1aa74fb 100644 --- a/moto/ec2instanceconnect/models.py +++ b/moto/ec2instanceconnect/models.py @@ -3,7 +3,7 @@ from moto.core import BaseBackend, BackendDict 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} ) diff --git a/moto/ec2instanceconnect/responses.py b/moto/ec2instanceconnect/responses.py index 9e76d5510..b01d03bb2 100644 --- a/moto/ec2instanceconnect/responses.py +++ b/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() diff --git a/moto/es/exceptions.py b/moto/es/exceptions.py index 31fb77c9e..23c298b08 100644 --- a/moto/es/exceptions.py +++ b/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) diff --git a/moto/es/models.py b/moto/es/models.py index 6c7e0c141..eda2e915f 100644 --- a/moto/es/models.py +++ b/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 @@ from .exceptions import DomainNotFound 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 @@ class Domain(BaseModel): 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 @@ class Domain(BaseModel): 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 @@ class ElasticsearchServiceBackend(BaseBackend): 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. diff --git a/moto/es/responses.py b/moto/es/responses.py index e61e3e77b..3fb4f7e6d 100644 --- a/moto/es/responses.py +++ b/moto/es/responses.py @@ -1,38 +1,40 @@ 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": @@ -40,7 +42,7 @@ class ElasticsearchServiceResponse(BaseResponse): 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 @@ class ElasticsearchServiceResponse(BaseResponse): ) 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 @@ class ElasticsearchServiceResponse(BaseResponse): ) 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}) diff --git a/setup.cfg b/setup.cfg index 1b31e3e43..4b0e2af7d 100644 --- a/setup.cfg +++ b/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