From e2d35824712e601dde86c023fbaa49d080e50d45 Mon Sep 17 00:00:00 2001 From: Bert Blommers Date: Sat, 29 Apr 2023 10:40:11 +0000 Subject: [PATCH] Techdebt: MyPy U (#6265) --- moto/cognitoidp/models.py | 12 +++--- moto/databrew/models.py | 10 ++--- moto/dax/models.py | 2 +- moto/ds/models.py | 4 +- moto/events/models.py | 4 +- moto/glue/models.py | 10 ++--- moto/iot/models.py | 2 +- moto/kinesis/models.py | 2 +- moto/logs/models.py | 2 +- moto/organizations/models.py | 6 +-- moto/route53/models.py | 2 +- moto/route53resolver/models.py | 10 ++--- moto/sagemaker/models.py | 8 ++-- moto/ssoadmin/models.py | 2 +- moto/stepfunctions/models.py | 4 +- moto/utilities/aws_headers.py | 24 +++++++---- moto/utilities/distutils_version.py | 36 ++++++++-------- moto/utilities/docker_utilities.py | 11 +++-- moto/utilities/paginator.py | 57 ++++++++++++++------------ moto/utilities/tagging_service.py | 12 +++--- moto/utilities/utils.py | 31 +++++++------- setup.cfg | 2 +- tests/test_utilities/test_paginator.py | 18 ++++---- 23 files changed, 142 insertions(+), 129 deletions(-) diff --git a/moto/cognitoidp/models.py b/moto/cognitoidp/models.py index d0e19a2e3..b8cbb99bb 100644 --- a/moto/cognitoidp/models.py +++ b/moto/cognitoidp/models.py @@ -961,7 +961,7 @@ class CognitoIdpBackend(BaseBackend): "MfaConfiguration": user_pool.mfa_config, } - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_user_pools(self) -> List[CognitoIdpUserPool]: # type: ignore[misc] return list(self.user_pools.values()) @@ -1034,7 +1034,7 @@ class CognitoIdpBackend(BaseBackend): user_pool.clients[user_pool_client.id] = user_pool_client return user_pool_client - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_user_pool_clients(self, user_pool_id: str) -> List[CognitoIdpUserPoolClient]: # type: ignore[misc] user_pool = self.describe_user_pool(user_pool_id) @@ -1081,7 +1081,7 @@ class CognitoIdpBackend(BaseBackend): user_pool.identity_providers[name] = identity_provider return identity_provider - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_identity_providers(self, user_pool_id: str) -> List[CognitoIdpIdentityProvider]: # type: ignore[misc] user_pool = self.describe_user_pool(user_pool_id) @@ -1147,7 +1147,7 @@ class CognitoIdpBackend(BaseBackend): return user_pool.groups[group_name] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_groups(self, user_pool_id: str) -> List[CognitoIdpGroup]: # type: ignore[misc] user_pool = self.describe_user_pool(user_pool_id) @@ -1188,7 +1188,7 @@ class CognitoIdpBackend(BaseBackend): group.users.add(user) user.groups.add(group) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_users_in_group(self, user_pool_id: str, group_name: str) -> List[CognitoIdpUser]: # type: ignore[misc] user_pool = self.describe_user_pool(user_pool_id) group = self.get_group(user_pool_id, group_name) @@ -1324,7 +1324,7 @@ class CognitoIdpBackend(BaseBackend): return user raise NotAuthorizedError("Invalid token") - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_users(self, user_pool_id: str) -> List[CognitoIdpUser]: # type: ignore[misc] user_pool = self.describe_user_pool(user_pool_id) diff --git a/moto/databrew/models.py b/moto/databrew/models.py index e4cd4a440..ba1d5e642 100644 --- a/moto/databrew/models.py +++ b/moto/databrew/models.py @@ -134,7 +134,7 @@ class DataBrewBackend(BaseBackend): recipe = self.recipes[recipe_name] recipe.update(recipe_description, recipe_steps) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_recipes(self, recipe_version: Optional[str] = None) -> List["FakeRecipeVersion"]: # type: ignore[misc] # https://docs.aws.amazon.com/databrew/latest/dg/API_ListRecipes.html if recipe_version == FakeRecipe.LATEST_WORKING: @@ -149,7 +149,7 @@ class DataBrewBackend(BaseBackend): recipes = [getattr(self.recipes[key], version) for key in self.recipes] return [r for r in recipes if r is not None] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_recipe_versions(self, recipe_name: str) -> List["FakeRecipeVersion"]: # type: ignore[misc] # https://docs.aws.amazon.com/databrew/latest/dg/API_ListRecipeVersions.html self.validate_length(recipe_name, "name", 255) @@ -253,7 +253,7 @@ class DataBrewBackend(BaseBackend): raise RulesetNotFoundException(ruleset_name) return self.rulesets[ruleset_name] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_rulesets(self) -> List["FakeRuleset"]: # type: ignore[misc] return list(self.rulesets.values()) @@ -288,7 +288,7 @@ class DataBrewBackend(BaseBackend): self.datasets[dataset_name] = dataset return dataset - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_datasets(self) -> List["FakeDataset"]: # type: ignore[misc] return list(self.datasets.values()) @@ -405,7 +405,7 @@ class DataBrewBackend(BaseBackend): # https://docs.aws.amazon.com/databrew/latest/dg/API_UpdateProfileJob.html return self.update_job(**kwargs) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_jobs(self, dataset_name: Optional[str] = None, project_name: Optional[str] = None) -> List["FakeJob"]: # type: ignore[misc] # https://docs.aws.amazon.com/databrew/latest/dg/API_ListJobs.html if dataset_name is not None: diff --git a/moto/dax/models.py b/moto/dax/models.py index 2a29194e5..f81387bd8 100644 --- a/moto/dax/models.py +++ b/moto/dax/models.py @@ -215,7 +215,7 @@ class DAXBackend(BaseBackend): self.clusters[cluster_name].delete() return self.clusters[cluster_name] - @paginate(PAGINATION_MODEL) + @paginate(PAGINATION_MODEL) # type: ignore[misc] def describe_clusters(self, cluster_names: Iterable[str]) -> List[DaxCluster]: # type: ignore[misc] clusters = self.clusters if not cluster_names: diff --git a/moto/ds/models.py b/moto/ds/models.py index 1d003f498..2dadfee01 100644 --- a/moto/ds/models.py +++ b/moto/ds/models.py @@ -476,7 +476,7 @@ class DirectoryServiceBackend(BaseBackend): directory = self.directories[directory_id] directory.enable_sso(True) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def describe_directories(self, directory_ids: Optional[List[str]] = None) -> List[Directory]: # type: ignore[misc] """Return info on all directories or directories with matching IDs.""" for directory_id in directory_ids or self.directories: @@ -530,7 +530,7 @@ class DirectoryServiceBackend(BaseBackend): self._validate_directory_id(resource_id) self.tagger.untag_resource_using_names(resource_id, tag_keys) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_tags_for_resource(self, resource_id: str) -> List[Dict[str, str]]: # type: ignore[misc] """List all tags on a directory.""" self._validate_directory_id(resource_id) diff --git a/moto/events/models.py b/moto/events/models.py index f2908c22e..767d51864 100644 --- a/moto/events/models.py +++ b/moto/events/models.py @@ -1164,7 +1164,7 @@ class EventsBackend(BaseBackend): return False - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_rule_names_by_target(self, target_arn: str, event_bus_arn: Optional[str]) -> List[Rule]: # type: ignore[misc] event_bus_name = self._normalize_event_bus_arn(event_bus_arn) event_bus = self._get_event_bus(event_bus_name) @@ -1177,7 +1177,7 @@ class EventsBackend(BaseBackend): return matching_rules - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_rules(self, prefix: Optional[str] = None, event_bus_arn: Optional[str] = None) -> List[Rule]: # type: ignore[misc] event_bus_name = self._normalize_event_bus_arn(event_bus_arn) event_bus = self._get_event_bus(event_bus_name) diff --git a/moto/glue/models.py b/moto/glue/models.py index 56e51a02a..6b288d650 100644 --- a/moto/glue/models.py +++ b/moto/glue/models.py @@ -320,7 +320,7 @@ class GlueBackend(BaseBackend): def get_crawlers(self) -> List["FakeCrawler"]: return [self.crawlers[key] for key in self.crawlers] if self.crawlers else [] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_crawlers(self) -> List["FakeCrawler"]: # type: ignore[misc] return [crawler for _, crawler in self.crawlers.items()] @@ -395,7 +395,7 @@ class GlueBackend(BaseBackend): except KeyError: raise JobNotFoundException(name) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def get_jobs(self) -> List["FakeJob"]: # type: ignore return [job for _, job in self.jobs.items()] @@ -407,7 +407,7 @@ class GlueBackend(BaseBackend): job = self.get_job(name) return job.get_job_run(run_id) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_jobs(self) -> List["FakeJob"]: # type: ignore return [job for _, job in self.jobs.items()] @@ -812,7 +812,7 @@ class GlueBackend(BaseBackend): trigger = self.get_trigger(name) trigger.stop_trigger() - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def get_triggers(self, dependent_job_name: str) -> List["FakeTrigger"]: # type: ignore if dependent_job_name: triggers = [] @@ -826,7 +826,7 @@ class GlueBackend(BaseBackend): return list(self.triggers.values()) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_triggers(self, dependent_job_name: str) -> List["FakeTrigger"]: # type: ignore if dependent_job_name: triggers = [] diff --git a/moto/iot/models.py b/moto/iot/models.py index 69715bf14..33f3664c6 100644 --- a/moto/iot/models.py +++ b/moto/iot/models.py @@ -1719,7 +1719,7 @@ class IoTBackend(BaseBackend): return job_executions, next_token - @paginate(PAGINATION_MODEL) + @paginate(PAGINATION_MODEL) # type: ignore[misc] def list_job_executions_for_thing(self, thing_name: str, status: Optional[str]) -> List[Dict[str, Any]]: # type: ignore[misc] job_executions = [ self.job_executions[je].to_dict() diff --git a/moto/kinesis/models.py b/moto/kinesis/models.py index 0d6e7463b..171bad3e2 100644 --- a/moto/kinesis/models.py +++ b/moto/kinesis/models.py @@ -783,7 +783,7 @@ class KinesisBackend(BaseBackend): return current_shard_count - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_shards(self, stream_arn: Optional[str], stream_name: Optional[str]) -> List[Dict[str, Any]]: # type: ignore stream = self.describe_stream(stream_arn=stream_arn, stream_name=stream_name) shards = sorted(stream.shards.values(), key=lambda x: x.shard_id) diff --git a/moto/logs/models.py b/moto/logs/models.py index 09ed64f1a..f4a76433e 100644 --- a/moto/logs/models.py +++ b/moto/logs/models.py @@ -695,7 +695,7 @@ class LogsBackend(BaseBackend): raise ResourceNotFoundException() del self.groups[log_group_name] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def describe_log_groups(self, log_group_name_prefix: Optional[str] = None) -> List[Dict[str, Any]]: # type: ignore[misc] groups = [ group.to_describe_dict() diff --git a/moto/organizations/models.py b/moto/organizations/models.py index d40230184..80362a3d6 100644 --- a/moto/organizations/models.py +++ b/moto/organizations/models.py @@ -441,7 +441,7 @@ class OrganizationsBackend(BaseBackend): ou = self.get_organizational_unit_by_id(kwargs["OrganizationalUnitId"]) return ou.describe() - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_organizational_units_for_parent(self, **kwargs: Any) -> List[Dict[str, Any]]: # type: ignore parent_id = self.validate_parent_id(kwargs["parent_id"]) return [ @@ -515,12 +515,12 @@ class OrganizationsBackend(BaseBackend): next_token = str(len(accounts_resp)) return dict(CreateAccountStatuses=accounts_resp, NextToken=next_token) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_accounts(self) -> List[FakeAccount]: # type: ignore accounts = [account.describe() for account in self.accounts] return sorted(accounts, key=lambda x: x["JoinedTimestamp"]) # type: ignore - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_accounts_for_parent(self, **kwargs: Any) -> Any: # type: ignore parent_id = self.validate_parent_id(kwargs["parent_id"]) accounts = [ diff --git a/moto/route53/models.py b/moto/route53/models.py index 2e6cc8de2..81f9cb77d 100644 --- a/moto/route53/models.py +++ b/moto/route53/models.py @@ -877,7 +877,7 @@ class Route53Backend(BaseBackend): raise NoSuchQueryLoggingConfig() return self.query_logging_configs[query_logging_config_id] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_query_logging_configs(self, hosted_zone_id: Optional[str] = None) -> List[QueryLoggingConfig]: # type: ignore """Return a list of query logging configs.""" if hosted_zone_id: diff --git a/moto/route53resolver/models.py b/moto/route53resolver/models.py index 2deee1383..8c3fed139 100644 --- a/moto/route53resolver/models.py +++ b/moto/route53resolver/models.py @@ -754,7 +754,7 @@ class Route53ResolverBackend(BaseBackend): ) return self.resolver_rule_associations[resolver_rule_association_id] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_resolver_endpoint_ip_addresses(self, resolver_endpoint_id: str) -> List[Dict[str, Any]]: # type: ignore[misc] self._validate_resolver_endpoint_id(resolver_endpoint_id) endpoint = self.resolver_endpoints[resolver_endpoint_id] @@ -813,7 +813,7 @@ class Route53ResolverBackend(BaseBackend): return True - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_resolver_endpoints(self, filters: Any) -> List[ResolverEndpoint]: # type: ignore[misc] if not filters: filters = [] @@ -827,7 +827,7 @@ class Route53ResolverBackend(BaseBackend): endpoints.append(endpoint) return endpoints - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_resolver_rules(self, filters: Any) -> List[ResolverRule]: # type: ignore[misc] if not filters: filters = [] @@ -841,7 +841,7 @@ class Route53ResolverBackend(BaseBackend): rules.append(rule) return rules - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_resolver_rule_associations(self, filters: Any) -> List[ResolverRuleAssociation]: # type: ignore[misc] if not filters: filters = [] @@ -869,7 +869,7 @@ class Route53ResolverBackend(BaseBackend): f"Resolver endpoint with ID '{resource_arn}' does not exist" ) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_tags_for_resource(self, resource_arn: str) -> Optional[List[Dict[str, str]]]: # type: ignore[misc] self._matched_arn(resource_arn) return self.tagger.list_tags_for_resource(resource_arn).get("Tags") diff --git a/moto/sagemaker/models.py b/moto/sagemaker/models.py index cab49ff18..74cf00657 100644 --- a/moto/sagemaker/models.py +++ b/moto/sagemaker/models.py @@ -1343,7 +1343,7 @@ class SageMakerModelBackend(BaseBackend): resource.tags.extend(tags) return resource.tags - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_tags(self, arn: str) -> List[Dict[str, str]]: # type: ignore[misc] resource = self._get_resource_from_arn(arn) return resource.tags @@ -1352,7 +1352,7 @@ class SageMakerModelBackend(BaseBackend): resource = self._get_resource_from_arn(arn) resource.tags = [tag for tag in resource.tags if tag["Key"] not in tag_keys] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_experiments(self) -> List["FakeExperiment"]: # type: ignore[misc] return list(self.experiments.values()) @@ -1522,7 +1522,7 @@ class SageMakerModelBackend(BaseBackend): message=f"Could not find trial configuration '{arn}'." ) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_trials(self, experiment_name: Optional[str] = None, trial_component_name: Optional[str] = None) -> List["FakeTrial"]: # type: ignore[misc] trials_fetched = list(self.trials.values()) @@ -1581,7 +1581,7 @@ class SageMakerModelBackend(BaseBackend): ) -> None: self.trial_components[trial_component_name].update(details_json) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_trial_components(self, trial_name: Optional[str] = None) -> List["FakeTrialComponent"]: # type: ignore[misc] trial_components_fetched = list(self.trial_components.values()) diff --git a/moto/ssoadmin/models.py b/moto/ssoadmin/models.py index d83c96147..a132ef452 100644 --- a/moto/ssoadmin/models.py +++ b/moto/ssoadmin/models.py @@ -249,7 +249,7 @@ class SSOAdminBackend(BaseBackend): return permission_set raise ResourceNotFound - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_permission_sets(self, instance_arn: str) -> List[PermissionSet]: # type: ignore[misc] permission_sets = [] for permission_set in self.permission_sets: diff --git a/moto/stepfunctions/models.py b/moto/stepfunctions/models.py index 5628c95dd..3df09e648 100644 --- a/moto/stepfunctions/models.py +++ b/moto/stepfunctions/models.py @@ -500,7 +500,7 @@ class StepFunctionBackend(BaseBackend): self.state_machines.append(state_machine) return state_machine - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_state_machines(self) -> Iterable[StateMachine]: # type: ignore[misc] return sorted(self.state_machines, key=lambda x: x.creation_date) @@ -546,7 +546,7 @@ class StepFunctionBackend(BaseBackend): state_machine = self._get_state_machine_for_execution(execution_arn) return state_machine.stop_execution(execution_arn) - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def list_executions(self, state_machine_arn: str, status_filter: Optional[str] = None) -> Iterable[Execution]: # type: ignore[misc] """ The status of every execution is set to 'RUNNING' by default. diff --git a/moto/utilities/aws_headers.py b/moto/utilities/aws_headers.py index f3366e499..5f7350e2b 100644 --- a/moto/utilities/aws_headers.py +++ b/moto/utilities/aws_headers.py @@ -1,5 +1,10 @@ from functools import wraps -from typing import Any, Callable, TypeVar +from typing import Any, Callable, Dict, TypeVar, Optional, TYPE_CHECKING + +if TYPE_CHECKING: + from typing_extensions import Protocol +else: + Protocol = object import binascii import re @@ -9,7 +14,12 @@ from moto.moto_api._internal import mock_random as random TypeDec = TypeVar("TypeDec", bound=Callable[..., Any]) -def gen_amz_crc32(response, headerdict=None): +class GenericFunction(Protocol): + def __call__(self, *args: Any, **kwargs: Any) -> Any: + ... + + +def gen_amz_crc32(response: Any, headerdict: Optional[Dict[str, Any]] = None) -> int: if not isinstance(response, bytes): response = response.encode("utf-8") @@ -21,7 +31,7 @@ def gen_amz_crc32(response, headerdict=None): return crc -def gen_amzn_requestid_long(headerdict=None): +def gen_amzn_requestid_long(headerdict: Optional[Dict[str, Any]] = None) -> str: req_id = random.get_random_string(length=52) if headerdict is not None and isinstance(headerdict, dict): @@ -30,9 +40,9 @@ def gen_amzn_requestid_long(headerdict=None): return req_id -def amz_crc32(f: TypeDec) -> TypeDec: +def amz_crc32(f: TypeDec) -> GenericFunction: @wraps(f) - def _wrapper(*args: Any, **kwargs: Any) -> Any: + def _wrapper(*args: Any, **kwargs: Any) -> Any: # type: ignore[misc] response = f(*args, **kwargs) headers = {} @@ -58,9 +68,9 @@ def amz_crc32(f: TypeDec) -> TypeDec: return _wrapper -def amzn_request_id(f: TypeDec) -> TypeDec: +def amzn_request_id(f: TypeDec) -> GenericFunction: @wraps(f) - def _wrapper(*args: Any, **kwargs: Any) -> Any: + def _wrapper(*args: Any, **kwargs: Any) -> Any: # type: ignore[misc] response = f(*args, **kwargs) headers = {} diff --git a/moto/utilities/distutils_version.py b/moto/utilities/distutils_version.py index 12d6702bd..a1c90766c 100644 --- a/moto/utilities/distutils_version.py +++ b/moto/utilities/distutils_version.py @@ -30,7 +30,7 @@ Every version number class implements the following interface: """ import re -from typing import Optional +from typing import Any, Optional class Version: @@ -40,39 +40,39 @@ class Version: rich comparisons to _cmp. """ - def __init__(self, vstring=None): + def __init__(self, vstring: Optional[str] = None): if vstring: - self.parse(vstring) + self.parse(vstring) # type: ignore[attr-defined] - def __repr__(self): + def __repr__(self) -> str: return f"{self.__class__.__name__} ('{self}')" - def __eq__(self, other): - c = self._cmp(other) + def __eq__(self, other: Any) -> bool: + c = self._cmp(other) # type: ignore[attr-defined] if c is NotImplemented: return c return c == 0 - def __lt__(self, other): - c = self._cmp(other) + def __lt__(self, other: Any) -> bool: + c = self._cmp(other) # type: ignore[attr-defined] if c is NotImplemented: return c return c < 0 - def __le__(self, other): - c = self._cmp(other) + def __le__(self, other: Any) -> bool: + c = self._cmp(other) # type: ignore[attr-defined] if c is NotImplemented: return c return c <= 0 - def __gt__(self, other): - c = self._cmp(other) + def __gt__(self, other: Any) -> bool: + c = self._cmp(other) # type: ignore[attr-defined] if c is NotImplemented: return c return c > 0 - def __ge__(self, other): - c = self._cmp(other) + def __ge__(self, other: Any) -> bool: + c = self._cmp(other) # type: ignore[attr-defined] if c is NotImplemented: return c return c >= 0 @@ -199,7 +199,7 @@ class LooseVersion(Version): if vstring: self.parse(vstring) - def parse(self, vstring): + def parse(self, vstring: str) -> None: # I've given up on thinking I can reconstruct the version string # from the parsed tuple -- so I just store the string here for # use by __str__ @@ -213,13 +213,13 @@ class LooseVersion(Version): self.version = components - def __str__(self): + def __str__(self) -> str: return self.vstring - def __repr__(self): + def __repr__(self) -> str: return f"LooseVersion ('{self}')" - def _cmp(self, other): + def _cmp(self, other: Any) -> int: # type: ignore[return] if isinstance(other, str): other = LooseVersion(other) diff --git a/moto/utilities/docker_utilities.py b/moto/utilities/docker_utilities.py index df2987985..9a0f28087 100644 --- a/moto/utilities/docker_utilities.py +++ b/moto/utilities/docker_utilities.py @@ -1,9 +1,12 @@ import functools import requests.adapters -from typing import Tuple +from typing import Any, Tuple, TYPE_CHECKING from moto import settings +if TYPE_CHECKING: + from docker import DockerClient + _orig_adapter_send = requests.adapters.HTTPAdapter.send @@ -13,7 +16,7 @@ class DockerModel: self.__docker_client = None @property - def docker_client(self): + def docker_client(self) -> "DockerClient": # type: ignore if self.__docker_client is None: # We should only initiate the Docker Client at runtime. # The docker.from_env() call will fall if Docker is not running @@ -26,11 +29,11 @@ class DockerModel: if requests.adapters.HTTPAdapter.send != _orig_adapter_send: _orig_get_adapter = self.docker_client.api.get_adapter - def replace_adapter_send(*args, **kwargs): + def replace_adapter_send(*args: Any, **kwargs: Any) -> Any: adapter = _orig_get_adapter(*args, **kwargs) if isinstance(adapter, requests.adapters.HTTPAdapter): - adapter.send = functools.partial(_orig_adapter_send, adapter) + adapter.send = functools.partial(_orig_adapter_send, adapter) # type: ignore return adapter self.docker_client.api.get_adapter = replace_adapter_send diff --git a/moto/utilities/paginator.py b/moto/utilities/paginator.py index 2a07f763e..ce411e49d 100644 --- a/moto/utilities/paginator.py +++ b/moto/utilities/paginator.py @@ -2,19 +2,23 @@ import inspect from copy import deepcopy from functools import wraps -from typing import Dict, Any, Callable +from typing import Any, Dict, List, Tuple, Optional from botocore.paginate import TokenDecoder, TokenEncoder from moto.core.exceptions import InvalidToken -def paginate( - pagination_model: Dict[str, Any], original_function: Callable = None -) -> Callable: - def pagination_decorator(func): +# This should be typed using ParamSpec +# https://stackoverflow.com/a/70591060/13245310 +# This currently does not work for our usecase +# I believe this could be fixed after https://github.com/python/mypy/pull/14903 is accepted + + +def paginate(pagination_model: Dict[str, Any]) -> Any: + def pagination_decorator(func: Any) -> Any: @wraps(func) - def pagination_wrapper(*args, **kwargs): + def pagination_wrapper(*args: Any, **kwargs: Any) -> Any: # type: ignore method = func.__name__ model = pagination_model @@ -59,27 +63,26 @@ def paginate( return pagination_wrapper - if original_function: - return pagination_decorator(original_function) - return pagination_decorator -class Paginator(object): +class Paginator: def __init__( self, - max_results=None, - max_results_default=None, - starting_token=None, - unique_attribute=None, - param_values_to_check=None, - fail_on_invalid_token=True, + max_results: Any = None, + max_results_default: Any = None, + starting_token: Any = None, + unique_attribute: Any = None, + param_values_to_check: Any = None, + fail_on_invalid_token: bool = True, ): self._max_results = max_results if max_results else max_results_default self._starting_token = starting_token - self._unique_attributes = unique_attribute - if not isinstance(unique_attribute, list): - self._unique_attributes = [unique_attribute] + self._unique_attributes = ( + unique_attribute + if isinstance(unique_attribute, list) + else [unique_attribute] + ) self._param_values_to_check = param_values_to_check self._fail_on_invalid_token = fail_on_invalid_token self._token_encoder = TokenEncoder() @@ -87,7 +90,7 @@ class Paginator(object): self._param_checksum = self._calculate_parameter_checksum() self._parsed_token = self._parse_starting_token() - def _parse_starting_token(self): + def _parse_starting_token(self) -> Optional[Dict[str, Any]]: if self._starting_token is None: return None # The starting token is a dict passed as a base64 encoded string. @@ -101,7 +104,7 @@ class Paginator(object): raise InvalidToken(f"Input inconsistent with page token: {str(next_token)}") return next_token - def _raise_exception_if_required(self, token): + def _raise_exception_if_required(self, token: Optional[str]) -> None: if self._fail_on_invalid_token: if isinstance(self._fail_on_invalid_token, type): # we need to raise a custom exception @@ -115,8 +118,8 @@ class Paginator(object): raise self._fail_on_invalid_token() raise InvalidToken("Invalid token") - def _calculate_parameter_checksum(self): - def freeze(o): + def _calculate_parameter_checksum(self) -> int: + def freeze(o: Any) -> Any: if not o: return None if isinstance(o, dict): @@ -129,7 +132,7 @@ class Paginator(object): return hash(freeze(self._param_values_to_check)) - def _check_predicate(self, item): + def _check_predicate(self, item: Any) -> bool: if self._parsed_token is None: return False unique_attributes = self._parsed_token["uniqueAttributes"] @@ -140,8 +143,8 @@ class Paginator(object): return False return True - def _build_next_token(self, next_item): - token_dict = {} + def _build_next_token(self, next_item: Any) -> str: + token_dict: Dict[str, Any] = {} if self._param_checksum: token_dict["parameterChecksum"] = self._param_checksum range_keys = [] @@ -153,7 +156,7 @@ class Paginator(object): token_dict["uniqueAttributes"] = "|".join(range_keys) return self._token_encoder.encode(token_dict) - def paginate(self, results): + def paginate(self, results: List[Any]) -> Tuple[List[Any], Optional[str]]: index_start = 0 if self._starting_token: try: diff --git a/moto/utilities/tagging_service.py b/moto/utilities/tagging_service.py index 970d1795d..66d934084 100644 --- a/moto/utilities/tagging_service.py +++ b/moto/utilities/tagging_service.py @@ -12,7 +12,7 @@ class TaggingService: self.tag_name = tag_name self.key_name = key_name self.value_name = value_name - self.tags: Dict[str, str] = {} + self.tags: Dict[str, Dict[str, Optional[str]]] = {} def get_tag_dict_for_resource(self, arn: str) -> Dict[str, str]: """Return dict of key/value pairs vs. list of key/values dicts.""" @@ -20,7 +20,7 @@ class TaggingService: if self.has_tags(arn): for key, val in self.tags[arn].items(): result[key] = val - return result + return result # type: ignore def list_tags_for_resource(self, arn: str) -> Dict[str, List[Dict[str, str]]]: """Return list of tags inside dict with key of "tag_name". @@ -32,7 +32,7 @@ class TaggingService: if self.has_tags(arn): for key, val in self.tags[arn].items(): result.append({self.key_name: key, self.value_name: val}) - return {self.tag_name: result} + return {self.tag_name: result} # type: ignore def delete_all_tags_for_resource(self, arn: str) -> None: """Delete all tags associated with given ARN.""" @@ -88,7 +88,7 @@ class TaggingService: def extract_tag_names(self, tags: List[Dict[str, str]]) -> List[str]: """Return list of key names in list of 'tags' key/value dicts.""" - results = [] + results: List[str] = [] if len(tags) == 0: return results for tag in tags: @@ -96,9 +96,9 @@ class TaggingService: results.append(tag[self.key_name]) return results - def flatten_tag_list(self, tags: List[Dict[str, str]]) -> Dict[str, str]: + def flatten_tag_list(self, tags: List[Dict[str, str]]) -> Dict[str, Optional[str]]: """Return dict of key/value pairs with 'tag_name', 'value_name'.""" - result = {} + result: Dict[str, Optional[str]] = {} for tag in tags: if self.value_name in tag: result[tag[self.key_name]] = tag[self.value_name] diff --git a/moto/utilities/utils.py b/moto/utilities/utils.py index ec0188493..73891a2fd 100644 --- a/moto/utilities/utils.py +++ b/moto/utilities/utils.py @@ -2,8 +2,7 @@ import json import hashlib import pkgutil -from collections.abc import MutableMapping -from typing import Any, Dict, List, TypeVar, Tuple, Optional +from typing import Any, Dict, Iterator, List, TypeVar, Tuple, Optional, MutableMapping def str2bool(v: Any) -> Optional[bool]: @@ -20,16 +19,14 @@ def load_resource(package: str, resource: str) -> Any: Usage: load_resource(__name__, "resources/file.json") """ - resource = pkgutil.get_data(package, resource) - return json.loads(resource) + return json.loads(pkgutil.get_data(package, resource)) # type: ignore def load_resource_as_str(package: str, resource: str) -> str: - resource = pkgutil.get_data(package, resource) - return resource.decode("utf-8") + return pkgutil.get_data(package, resource).decode("utf-8") # type: ignore -def merge_multiple_dicts(*args: Any) -> Dict[str, any]: +def merge_multiple_dicts(*args: Any) -> Dict[str, Any]: result = {} for d in args: result.update(d) @@ -68,36 +65,36 @@ def md5_hash(data: Any = None) -> Any: """ args = (data,) if data else () try: - return hashlib.md5(*args, usedforsecurity=False) + return hashlib.md5(*args, usedforsecurity=False) # type: ignore except TypeError: # The usedforsecurity-parameter is only available as of Python 3.9 return hashlib.md5(*args) -class LowercaseDict(MutableMapping): +class LowercaseDict(MutableMapping[str, Any]): """A dictionary that lowercases all keys""" def __init__(self, *args: Any, **kwargs: Any): - self.store = dict() + self.store: Dict[str, Any] = dict() self.update(dict(*args, **kwargs)) # use the free update to set keys - def __getitem__(self, key): + def __getitem__(self, key: str) -> Any: return self.store[self._keytransform(key)] - def __setitem__(self, key, value): + def __setitem__(self, key: str, value: Any) -> None: self.store[self._keytransform(key)] = value - def __delitem__(self, key): + def __delitem__(self, key: str) -> None: del self.store[self._keytransform(key)] - def __iter__(self): + def __iter__(self) -> Iterator[Any]: return iter(self.store) - def __len__(self): + def __len__(self) -> int: return len(self.store) - def __repr__(self): + def __repr__(self) -> str: return str(self.store) - def _keytransform(self, key): + def _keytransform(self, key: str) -> str: return key.lower() diff --git a/setup.cfg b/setup.cfg index 32597d466..8fe537bee 100644 --- a/setup.cfg +++ b/setup.cfg @@ -239,7 +239,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/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/q*,moto/r*,moto/s* +files= moto/a*,moto/b*,moto/c*,moto/d*,moto/e*,moto/f*,moto/g*,moto/i*,moto/k*,moto/l*,moto/m*,moto/n*,moto/o*,moto/p*,moto/q*,moto/r*,moto/s*,moto/u* show_column_numbers=True show_error_codes = True disable_error_code=abstract diff --git a/tests/test_utilities/test_paginator.py b/tests/test_utilities/test_paginator.py index 2ece17673..ed612e6a4 100644 --- a/tests/test_utilities/test_paginator.py +++ b/tests/test_utilities/test_paginator.py @@ -163,41 +163,41 @@ class TestDecorator(unittest.TestCase): "method_with_list_as_kwarg": {"limit_default": 1, "unique_attribute": "name"}, } - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def method_returning_dict(self): return results - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def method_returning_instances(self): return model_results - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def method_without_configuration(self): return results - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def method_returning_args(self, *args, **kwargs): return [*args] + [(k, v) for k, v in kwargs.items()] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def method_expecting_token_as_kwarg(self, custom_token=None): self.custom_token = custom_token return [{"name": "item1"}, {"name": "item2"}] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def method_expecting_limit_as_kwarg(self, custom_limit): self.custom_limit = custom_limit return [{"name": "item1"}, {"name": "item2"}] - @paginate(pagination_model=PAGINATION_MODEL) + @paginate(pagination_model=PAGINATION_MODEL) # type: ignore[misc] def method_with_list_as_kwarg(self, resources=[]): return resources or results - @paginate(PAGINATION_MODEL) + @paginate(PAGINATION_MODEL) # type: ignore[misc] def method_specifying_invalidtoken_exception(self): return results - @paginate(PAGINATION_MODEL) + @paginate(PAGINATION_MODEL) # type: ignore[misc] def method_specifying_generic_invalidtoken_exception(self): return results