From 4d6271d1bdec1c5eec5789d7c6cc8ef35566d0c2 Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Fri, 24 Feb 2023 09:29:21 -0100 Subject: [PATCH] Techdebt: MyPy ElastiCache (#5970) --- moto/elasticache/exceptions.py | 21 +++++++++---------- moto/elasticache/models.py | 37 ++++++++++++++++++++-------------- moto/elasticache/responses.py | 22 ++++++++++---------- setup.cfg | 2 +- 4 files changed, 43 insertions(+), 39 deletions(-) diff --git a/moto/elasticache/exceptions.py b/moto/elasticache/exceptions.py index 8ce9ff14f..fe14cb050 100644 --- a/moto/elasticache/exceptions.py +++ b/moto/elasticache/exceptions.py @@ -1,3 +1,4 @@ +from typing import Any from moto.core.exceptions import RESTError EXCEPTION_RESPONSE = """ @@ -15,21 +16,20 @@ class ElastiCacheException(RESTError): code = 400 - def __init__(self, *args, **kwargs): + def __init__(self, code: str, message: str, **kwargs: Any): kwargs.setdefault("template", "ecerror") self.templates["ecerror"] = EXCEPTION_RESPONSE - super().__init__(*args, **kwargs) + super().__init__(code, message) class PasswordTooShort(ElastiCacheException): code = 404 - def __init__(self, **kwargs): + def __init__(self) -> None: super().__init__( "InvalidParameterValue", message="Passwords length must be between 16-128 characters.", - **kwargs, ) @@ -37,11 +37,10 @@ class PasswordRequired(ElastiCacheException): code = 404 - def __init__(self, **kwargs): + def __init__(self) -> None: super().__init__( "InvalidParameterValue", message="No password was provided. If you want to create/update the user without password, please use the NoPasswordRequired flag.", - **kwargs, ) @@ -49,15 +48,13 @@ class UserAlreadyExists(ElastiCacheException): code = 404 - def __init__(self, **kwargs): - super().__init__( - "UserAlreadyExists", message="User user1 already exists.", **kwargs - ) + def __init__(self) -> None: + super().__init__("UserAlreadyExists", message="User user1 already exists.") class UserNotFound(ElastiCacheException): code = 404 - def __init__(self, user_id, **kwargs): - super().__init__("UserNotFound", message=f"User {user_id} not found.", **kwargs) + def __init__(self, user_id: str): + super().__init__("UserNotFound", message=f"User {user_id} not found.") diff --git a/moto/elasticache/models.py b/moto/elasticache/models.py index 6b99ebab9..198d75dee 100644 --- a/moto/elasticache/models.py +++ b/moto/elasticache/models.py @@ -1,3 +1,4 @@ +from typing import List, Optional from moto.core import BaseBackend, BackendDict, BaseModel from .exceptions import UserAlreadyExists, UserNotFound @@ -6,14 +7,14 @@ from .exceptions import UserAlreadyExists, UserNotFound class User(BaseModel): def __init__( self, - account_id, - region, - user_id, - user_name, - access_string, - engine, - no_password_required, - passwords=None, + account_id: str, + region: str, + user_id: str, + user_name: str, + access_string: str, + engine: str, + no_password_required: bool, + passwords: Optional[List[str]] = None, ): self.id = user_id self.name = user_name @@ -23,7 +24,7 @@ class User(BaseModel): self.no_password_required = no_password_required self.status = "active" self.minimum_engine_version = "6.0" - self.usergroupids = [] + self.usergroupids: List[str] = [] self.region = region self.arn = f"arn:aws:elasticache:{self.region}:{account_id}:user:{self.id}" @@ -31,7 +32,7 @@ class User(BaseModel): class ElastiCacheBackend(BaseBackend): """Implementation of ElastiCache APIs.""" - def __init__(self, region_name, account_id): + def __init__(self, region_name: str, account_id: str): super().__init__(region_name, account_id) self.users = dict() self.users["default"] = User( @@ -45,8 +46,14 @@ class ElastiCacheBackend(BaseBackend): ) def create_user( - self, user_id, user_name, engine, passwords, access_string, no_password_required - ): + self, + user_id: str, + user_name: str, + engine: str, + passwords: List[str], + access_string: str, + no_password_required: bool, + ) -> User: if user_id in self.users: raise UserAlreadyExists user = User( @@ -62,7 +69,7 @@ class ElastiCacheBackend(BaseBackend): self.users[user_id] = user return user - def delete_user(self, user_id): + def delete_user(self, user_id: str) -> User: if user_id in self.users: user = self.users[user_id] if user.status == "active": @@ -70,7 +77,7 @@ class ElastiCacheBackend(BaseBackend): return user raise UserNotFound(user_id) - def describe_users(self, user_id): + def describe_users(self, user_id: Optional[str]) -> List[User]: """ Only the `user_id` parameter is currently supported. Pagination is not yet implemented. @@ -83,7 +90,7 @@ class ElastiCacheBackend(BaseBackend): return [user] else: raise UserNotFound(user_id) - return self.users.values() + return list(self.users.values()) elasticache_backends = BackendDict(ElastiCacheBackend, "elasticache") diff --git a/moto/elasticache/responses.py b/moto/elasticache/responses.py index 087ad6814..273fa002d 100644 --- a/moto/elasticache/responses.py +++ b/moto/elasticache/responses.py @@ -1,20 +1,20 @@ from moto.core.responses import BaseResponse from .exceptions import PasswordTooShort, PasswordRequired -from .models import elasticache_backends +from .models import elasticache_backends, ElastiCacheBackend class ElastiCacheResponse(BaseResponse): """Handler for ElastiCache requests and responses.""" - def __init__(self): + def __init__(self) -> None: super().__init__(service_name="elasticache") @property - def elasticache_backend(self): + def elasticache_backend(self) -> ElastiCacheBackend: """Return backend instance specific for this region.""" return elasticache_backends[self.current_account][self.region] - def create_user(self): + def create_user(self) -> str: params = self._get_params() user_id = params.get("UserId") user_name = params.get("UserName") @@ -28,24 +28,24 @@ class ElastiCacheResponse(BaseResponse): raise PasswordTooShort access_string = params.get("AccessString") user = self.elasticache_backend.create_user( - user_id=user_id, - user_name=user_name, - engine=engine, + user_id=user_id, # type: ignore[arg-type] + user_name=user_name, # type: ignore[arg-type] + engine=engine, # type: ignore[arg-type] passwords=passwords, - access_string=access_string, + access_string=access_string, # type: ignore[arg-type] no_password_required=no_password_required, ) template = self.response_template(CREATE_USER_TEMPLATE) return template.render(user=user) - def delete_user(self): + def delete_user(self) -> str: params = self._get_params() user_id = params.get("UserId") - user = self.elasticache_backend.delete_user(user_id=user_id) + user = self.elasticache_backend.delete_user(user_id=user_id) # type: ignore[arg-type] template = self.response_template(DELETE_USER_TEMPLATE) return template.render(user=user) - def describe_users(self): + def describe_users(self) -> str: params = self._get_params() user_id = params.get("UserId") users = self.elasticache_backend.describe_users(user_id=user_id) diff --git a/setup.cfg b/setup.cfg index 228066350..15ba29675 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/ec2instanceconnect,moto/ecr,moto/ecs,moto/efs,moto/eks,moto/es,moto/moto_api +files= moto/a*,moto/b*,moto/c*,moto/d*,moto/ebs/,moto/ec2,moto/ec2instanceconnect,moto/ecr,moto/ecs,moto/efs,moto/eks,moto/elasticache,moto/es,moto/moto_api show_column_numbers=True show_error_codes = True disable_error_code=abstract