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

Techdebt: MyPy U (#6265)

Browse Source
This commit is contained in:
Bert Blommers 2023-04-29 10:40:11 +00:00 committed by GitHub
parent 2cd773fe95
commit e2d3582471
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
23 changed files with 142 additions and 129 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

12
moto/cognitoidp/models.py
View File

@ -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)

10
moto/databrew/models.py
View File

@ -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:

2
moto/dax/models.py
View File

@ -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:

4
moto/ds/models.py
View File

@ -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)

4
moto/events/models.py
View File

@ -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)

10
moto/glue/models.py
View File

@ -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 = []

2
moto/iot/models.py
View File

@ -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()

2
moto/kinesis/models.py
View File

@ -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)

2
moto/logs/models.py
View File

@ -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()

6
moto/organizations/models.py
View File

@ -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 = [

2
moto/route53/models.py
View File

@ -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:

10
moto/route53resolver/models.py
View File

@ -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")

8
moto/sagemaker/models.py
View File

@ -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())

2
moto/ssoadmin/models.py
View File

@ -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:

4
moto/stepfunctions/models.py
View File

@ -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.

24
moto/utilities/aws_headers.py
View File

@ -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 = {}

36
moto/utilities/distutils_version.py
View File

@ -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)

11
moto/utilities/docker_utilities.py
View File

@ -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

57
moto/utilities/paginator.py
View File

@ -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:

12
moto/utilities/tagging_service.py
View File

@ -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]

31
moto/utilities/utils.py
View File

@ -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()

2
setup.cfg
View File

@ -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

18
tests/test_utilities/test_paginator.py
View File

@ -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