Yasuke/moto
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

TechDebt: MyPy Comprehend (#5630)

Browse Source
This commit is contained in:
Bert Blommers 2022-11-02 21:45:36 -01:00 committed by GitHub
parent ecafc89153
commit d3b16b3052
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
4 changed files with 49 additions and 45 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
moto/comprehend/exceptions.py
View File

@ -3,7 +3,7 @@ from moto.core.exceptions import JsonRESTError
class ResourceNotFound(JsonRESTError):
def __init__(self):
def __init__(self) -> None:
super().__init__(
"ResourceNotFoundException",
"RESOURCE_NOT_FOUND: Could not find specified resource.",

68
moto/comprehend/models.py
View File

@ -4,23 +4,23 @@ from moto.core import BaseBackend, BaseModel
from moto.core.utils import BackendDict
from moto.utilities.tagging_service import TaggingService
from .exceptions import ResourceNotFound
from typing import Dict
from typing import Any, Dict, List, Iterable
class EntityRecognizer(BaseModel):
def __init__(
self,
region_name,
account_id,
language_code,
input_data_config,
data_access_role_arn,
version_name,
recognizer_name,
volume_kms_key_id,
vpc_config,
model_kms_key_id,
model_policy,
region_name: str,
account_id: str,
language_code: str,
input_data_config: Dict[str, Any],
data_access_role_arn: str,
version_name: str,
recognizer_name: str,
volume_kms_key_id: str,
vpc_config: Dict[str, List[str]],
model_kms_key_id: str,
model_policy: str,
):
self.name = recognizer_name
self.arn = f"arn:aws:comprehend:{region_name}:{account_id}:entity-recognizer/{recognizer_name}"
@ -36,7 +36,7 @@ class EntityRecognizer(BaseModel):
self.model_policy = model_policy
self.status = "TRAINED"
def to_dict(self):
def to_dict(self) -> Dict[str, Any]:
return {
"EntityRecognizerArn": self.arn,
"LanguageCode": self.language_code,
@ -54,12 +54,14 @@ class EntityRecognizer(BaseModel):
class ComprehendBackend(BaseBackend):
"""Implementation of Comprehend APIs."""
def __init__(self, region_name, account_id):
def __init__(self, region_name: str, account_id: str):
super().__init__(region_name, account_id)
self.recognizers: Dict[str, EntityRecognizer] = dict()
self.tagger = TaggingService()
def list_entity_recognizers(self, _filter):
def list_entity_recognizers(
self, _filter: Dict[str, Any]
) -> Iterable[EntityRecognizer]:
"""
Pagination is not yet implemented.
The following filters are not yet implemented: Status, SubmitTimeBefore, SubmitTimeAfter
@ -74,17 +76,17 @@ class ComprehendBackend(BaseBackend):
def create_entity_recognizer(
self,
recognizer_name,
version_name,
data_access_role_arn,
tags,
input_data_config,
language_code,
volume_kms_key_id,
vpc_config,
model_kms_key_id,
model_policy,
):
recognizer_name: str,
version_name: str,
data_access_role_arn: str,
tags: List[Dict[str, str]],
input_data_config: Dict[str, Any],
language_code: str,
volume_kms_key_id: str,
vpc_config: Dict[str, List[str]],
model_kms_key_id: str,
model_policy: str,
) -> str:
"""
The ClientRequestToken-parameter is not yet implemented
"""
@ -105,26 +107,28 @@ class ComprehendBackend(BaseBackend):
self.tagger.tag_resource(recognizer.arn, tags)
return recognizer.arn
def describe_entity_recognizer(self, entity_recognizer_arn) -> EntityRecognizer:
def describe_entity_recognizer(
self, entity_recognizer_arn: str
) -> EntityRecognizer:
if entity_recognizer_arn not in self.recognizers:
raise ResourceNotFound
return self.recognizers[entity_recognizer_arn]
def stop_training_entity_recognizer(self, entity_recognizer_arn):
def stop_training_entity_recognizer(self, entity_recognizer_arn: str) -> None:
recognizer = self.describe_entity_recognizer(entity_recognizer_arn)
if recognizer.status == "TRAINING":
recognizer.status = "STOP_REQUESTED"
def list_tags_for_resource(self, resource_arn):
def list_tags_for_resource(self, resource_arn: str) -> List[Dict[str, str]]:
return self.tagger.list_tags_for_resource(resource_arn)["Tags"]
def delete_entity_recognizer(self, entity_recognizer_arn):
def delete_entity_recognizer(self, entity_recognizer_arn: str) -> None:
self.recognizers.pop(entity_recognizer_arn, None)
def tag_resource(self, resource_arn, tags):
def tag_resource(self, resource_arn: str, tags: List[Dict[str, str]]) -> None:
self.tagger.tag_resource(resource_arn, tags)
def untag_resource(self, resource_arn, tag_keys):
def untag_resource(self, resource_arn: str, tag_keys: List[str]) -> None:
self.tagger.untag_resource_using_names(resource_arn, tag_keys)

22
moto/comprehend/responses.py
View File

@ -2,21 +2,21 @@
import json
from moto.core.responses import BaseResponse
from .models import comprehend_backends
from .models import comprehend_backends, ComprehendBackend
class ComprehendResponse(BaseResponse):
"""Handler for Comprehend requests and responses."""
def __init__(self):
def __init__(self) -> None:
super().__init__(service_name="comprehend")
@property
def comprehend_backend(self):
def comprehend_backend(self) -> ComprehendBackend:
"""Return backend instance specific for this region."""
return comprehend_backends[self.current_account][self.region]
def list_entity_recognizers(self):
def list_entity_recognizers(self) -> str:
params = json.loads(self.body)
_filter = params.get("Filter", {})
recognizers = self.comprehend_backend.list_entity_recognizers(_filter=_filter)
@ -24,7 +24,7 @@ class ComprehendResponse(BaseResponse):
dict(EntityRecognizerPropertiesList=[r.to_dict() for r in recognizers])
)
def create_entity_recognizer(self):
def create_entity_recognizer(self) -> str:
params = json.loads(self.body)
recognizer_name = params.get("RecognizerName")
version_name = params.get("VersionName")
@ -50,7 +50,7 @@ class ComprehendResponse(BaseResponse):
)
return json.dumps(dict(EntityRecognizerArn=entity_recognizer_arn))
def describe_entity_recognizer(self):
def describe_entity_recognizer(self) -> str:
params = json.loads(self.body)
entity_recognizer_arn = params.get("EntityRecognizerArn")
recognizer = self.comprehend_backend.describe_entity_recognizer(
@ -58,7 +58,7 @@ class ComprehendResponse(BaseResponse):
)
return json.dumps(dict(EntityRecognizerProperties=recognizer.to_dict()))
def stop_training_entity_recognizer(self):
def stop_training_entity_recognizer(self) -> str:
params = json.loads(self.body)
entity_recognizer_arn = params.get("EntityRecognizerArn")
self.comprehend_backend.stop_training_entity_recognizer(
@ -66,7 +66,7 @@ class ComprehendResponse(BaseResponse):
)
return json.dumps(dict())
def list_tags_for_resource(self):
def list_tags_for_resource(self) -> str:
params = json.loads(self.body)
resource_arn = params.get("ResourceArn")
tags = self.comprehend_backend.list_tags_for_resource(
@ -74,7 +74,7 @@ class ComprehendResponse(BaseResponse):
)
return json.dumps(dict(ResourceArn=resource_arn, Tags=tags))
def delete_entity_recognizer(self):
def delete_entity_recognizer(self) -> str:
params = json.loads(self.body)
entity_recognizer_arn = params.get("EntityRecognizerArn")
self.comprehend_backend.delete_entity_recognizer(
@ -82,14 +82,14 @@ class ComprehendResponse(BaseResponse):
)
return "{}"
def tag_resource(self):
def tag_resource(self) -> str:
params = json.loads(self.body)
resource_arn = params.get("ResourceArn")
tags = params.get("Tags")
self.comprehend_backend.tag_resource(resource_arn, tags)
return "{}"
def untag_resource(self):
def untag_resource(self) -> str:
params = json.loads(self.body)
resource_arn = params.get("ResourceArn")
tag_keys = params.get("TagKeys")

2
setup.cfg
View File

@ -18,7 +18,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/ce,moto/cloudformation,moto/cloudfront,moto/cloudtrail,moto/codebuild,moto/cloudwatch,moto/codepipeline,moto/codecommit,moto/cognito*
files= moto/a*,moto/b*,moto/ce,moto/cloudformation,moto/cloudfront,moto/cloudtrail,moto/codebuild,moto/cloudwatch,moto/codepipeline,moto/codecommit,moto/cognito*,moto/comprehend
show_column_numbers=True
show_error_codes = True
disable_error_code=abstract