From 8a16a6a86286983ea0c60591edbee729219b729f Mon Sep 17 00:00:00 2001 From: tungol Date: Fri, 1 Dec 2023 03:51:28 -0800 Subject: [PATCH] Techdebt: typing for backends (#7069) --- moto/awslambda/models.py | 4 +- moto/awslambda_simple/responses.py | 2 +- moto/backends.py | 535 ++++++++++++++++++++++++- moto/batch/models.py | 10 +- moto/batch_simple/responses.py | 2 +- moto/cloudformation/parsing.py | 9 +- moto/cloudformation/utils.py | 2 +- moto/cognitoidp/responses.py | 2 +- moto/core/__init__.py | 2 +- moto/core/base_backend.py | 21 +- moto/core/models.py | 8 +- moto/ds/models.py | 4 +- moto/ec2/models/__init__.py | 2 +- moto/ec2/models/elastic_block_store.py | 3 +- moto/ec2/models/instances.py | 2 +- moto/ec2/models/subnets.py | 5 +- moto/elbv2/models.py | 2 +- moto/events/models.py | 4 +- moto/events/notifications.py | 4 +- moto/iam/models.py | 2 +- moto/iot/models.py | 2 +- moto/logs/models.py | 2 +- moto/moto_server/werkzeug_app.py | 8 +- moto/rds/models.py | 26 +- moto/route53/models.py | 5 +- moto/route53resolver/models.py | 8 +- moto/s3/models.py | 2 +- moto/sagemaker/utils.py | 2 +- moto/ses/models.py | 3 +- moto/sqs/models.py | 15 +- moto/ssm/models.py | 4 +- setup.cfg | 2 +- 32 files changed, 617 insertions(+), 87 deletions(-) diff --git a/moto/awslambda/models.py b/moto/awslambda/models.py index 5d70c361f..f5be1747b 100644 --- a/moto/awslambda/models.py +++ b/moto/awslambda/models.py @@ -1281,7 +1281,7 @@ class EventSourceMapping(CloudFormationModel): properties = cloudformation_json["Properties"] event_source_uuid = original_resource.uuid lambda_backend = lambda_backends[account_id][region_name] - return lambda_backend.update_event_source_mapping(event_source_uuid, properties) + return lambda_backend.update_event_source_mapping(event_source_uuid, properties) # type: ignore[return-value] @classmethod def delete_from_cloudformation_json( # type: ignore[misc] @@ -1331,7 +1331,7 @@ class LambdaVersion(CloudFormationModel): properties = cloudformation_json["Properties"] function_name = properties["FunctionName"] func = lambda_backends[account_id][region_name].publish_function(function_name) - spec = {"Version": func.version} + spec = {"Version": func.version} # type: ignore[union-attr] return LambdaVersion(spec) diff --git a/moto/awslambda_simple/responses.py b/moto/awslambda_simple/responses.py index 72f138ce7..04d8c05a5 100644 --- a/moto/awslambda_simple/responses.py +++ b/moto/awslambda_simple/responses.py @@ -5,4 +5,4 @@ from .models import LambdaBackend, lambda_simple_backends class LambdaSimpleResponse(LambdaResponse): @property def backend(self) -> LambdaBackend: - return lambda_simple_backends[self.current_account][self.region] + return lambda_simple_backends[self.current_account][self.region] # type: ignore[return-value] diff --git a/moto/backends.py b/moto/backends.py index 186685ef5..18943a2a4 100644 --- a/moto/backends.py +++ b/moto/backends.py @@ -1,9 +1,140 @@ import importlib import sys -from typing import Iterable, Tuple +from typing import TYPE_CHECKING, Iterable, Tuple, Union, overload import moto -from moto.core import BackendDict + +if TYPE_CHECKING: + from typing_extensions import Literal + + from moto.acm.models import AWSCertificateManagerBackend + from moto.acmpca.models import ACMPCABackend + from moto.amp.models import PrometheusServiceBackend + from moto.apigateway.models import APIGatewayBackend + from moto.apigatewaymanagementapi.models import ApiGatewayManagementApiBackend + from moto.apigatewayv2.models import ApiGatewayV2Backend + from moto.appconfig.models import AppConfigBackend + from moto.applicationautoscaling.models import ApplicationAutoscalingBackend + from moto.appsync.models import AppSyncBackend + from moto.athena.models import AthenaBackend + from moto.autoscaling.models import AutoScalingBackend + from moto.awslambda.models import LambdaBackend + from moto.batch.models import BatchBackend + from moto.budgets.models import BudgetsBackend + from moto.ce.models import CostExplorerBackend + from moto.cloudformation.models import CloudFormationBackend + from moto.cloudfront.models import CloudFrontBackend + from moto.cloudtrail.models import CloudTrailBackend + from moto.cloudwatch.models import CloudWatchBackend + from moto.codebuild.models import CodeBuildBackend + from moto.codecommit.models import CodeCommitBackend + from moto.codepipeline.models import CodePipelineBackend + from moto.cognitoidentity.models import CognitoIdentityBackend + from moto.cognitoidp.models import CognitoIdpBackend + from moto.comprehend.models import ComprehendBackend + from moto.config.models import ConfigBackend + from moto.core import SERVICE_BACKEND, BackendDict, BaseBackend + from moto.databrew.models import DataBrewBackend + from moto.datapipeline.models import DataPipelineBackend + from moto.datasync.models import DataSyncBackend + from moto.dax.models import DAXBackend + from moto.dms.models import DatabaseMigrationServiceBackend + from moto.ds.models import DirectoryServiceBackend + from moto.dynamodb.models import DynamoDBBackend + from moto.dynamodb_v20111205.models import ( + DynamoDBBackend as DynamoDBBackend_v20111205, + ) + from moto.dynamodbstreams.models import DynamoDBStreamsBackend + from moto.ebs.models import EBSBackend + from moto.ec2.models import EC2Backend + from moto.ec2instanceconnect.models import Ec2InstanceConnectBackend + from moto.ecr.models import ECRBackend + from moto.ecs.models import EC2ContainerServiceBackend + from moto.efs.models import EFSBackend + from moto.eks.models import EKSBackend + from moto.elasticache.models import ElastiCacheBackend + from moto.elasticbeanstalk.models import EBBackend + from moto.elastictranscoder.models import ElasticTranscoderBackend + from moto.elb.models import ELBBackend + from moto.elbv2.models import ELBv2Backend + from moto.emr.models import ElasticMapReduceBackend + from moto.emrcontainers.models import EMRContainersBackend + from moto.emrserverless.models import EMRServerlessBackend + from moto.es.models import ElasticsearchServiceBackend + from moto.events.models import EventsBackend + from moto.firehose.models import FirehoseBackend + from moto.forecast.models import ForecastBackend + from moto.glacier.models import GlacierBackend + from moto.glue.models import GlueBackend + from moto.greengrass.models import GreengrassBackend + from moto.guardduty.models import GuardDutyBackend + from moto.iam.models import IAMBackend + from moto.identitystore.models import IdentityStoreBackend + from moto.inspector2.models import Inspector2Backend + from moto.instance_metadata.models import InstanceMetadataBackend + from moto.iot.models import IoTBackend + from moto.iotdata.models import IoTDataPlaneBackend + from moto.ivs.models import IVSBackend + from moto.kinesis.models import KinesisBackend + from moto.kinesisvideo.models import KinesisVideoBackend + from moto.kinesisvideoarchivedmedia.models import KinesisVideoArchivedMediaBackend + from moto.kms.models import KmsBackend + from moto.lakeformation.models import LakeFormationBackend + from moto.logs.models import LogsBackend + from moto.managedblockchain.models import ManagedBlockchainBackend + from moto.mediaconnect.models import MediaConnectBackend + from moto.medialive.models import MediaLiveBackend + from moto.mediapackage.models import MediaPackageBackend + from moto.mediastore.models import MediaStoreBackend + from moto.mediastoredata.models import MediaStoreDataBackend + from moto.meteringmarketplace.models import MeteringMarketplaceBackend + from moto.moto_api._internal.models import MotoAPIBackend + from moto.mq.models import MQBackend + from moto.neptune.models import NeptuneBackend + from moto.opensearch.models import OpenSearchServiceBackend + from moto.opsworks.models import OpsWorksBackend + from moto.organizations.models import OrganizationsBackend + from moto.personalize.models import PersonalizeBackend + from moto.pinpoint.models import PinpointBackend + from moto.polly.models import PollyBackend + from moto.quicksight.models import QuickSightBackend + from moto.ram.models import ResourceAccessManagerBackend + from moto.rds.models import RDSBackend + from moto.rdsdata.models import RDSDataServiceBackend + from moto.redshift.models import RedshiftBackend + from moto.redshiftdata.models import RedshiftDataAPIServiceBackend + from moto.rekognition.models import RekognitionBackend + from moto.resourcegroups.models import ResourceGroupsBackend + from moto.resourcegroupstaggingapi.models import ResourceGroupsTaggingAPIBackend + from moto.robomaker.models import RoboMakerBackend + from moto.route53.models import Route53Backend + from moto.route53resolver.models import Route53ResolverBackend + from moto.s3.models import S3Backend + from moto.s3control.models import S3ControlBackend + from moto.sagemaker.models import SageMakerModelBackend + from moto.sagemakerruntime.models import SageMakerRuntimeBackend + from moto.scheduler.models import EventBridgeSchedulerBackend + from moto.sdb.models import SimpleDBBackend + from moto.secretsmanager.models import SecretsManagerBackend + from moto.servicediscovery.models import ServiceDiscoveryBackend + from moto.servicequotas.models import ServiceQuotasBackend + from moto.ses.models import SESBackend + from moto.sesv2.models import SESV2Backend + from moto.signer.models import SignerBackend + from moto.sns.models import SNSBackend + from moto.sqs.models import SQSBackend + from moto.ssm.models import SimpleSystemManagerBackend + from moto.ssoadmin.models import SSOAdminBackend + from moto.stepfunctions.models import StepFunctionBackend + from moto.sts.models import STSBackend + from moto.support.models import SupportBackend + from moto.swf.models import SWFBackend + from moto.textract.models import TextractBackend + from moto.timestreamwrite.models import TimestreamWriteBackend + from moto.transcribe.models import TranscribeBackend + from moto.wafv2.models import WAFV2Backend + from moto.xray.models import XRayBackend + decorators = [d for d in dir(moto) if d.startswith("mock_") and not d == "mock_all"] decorator_functions = [getattr(moto, f) for f in decorators] @@ -13,24 +144,157 @@ BACKENDS["moto_api"] = ("moto_api._internal", "moto_api_backends") BACKENDS["instance_metadata"] = ("instance_metadata", "instance_metadata_backends") BACKENDS["s3bucket_path"] = ("s3", "s3_backends") +# There's a similar Union that we could import from boto3-stubs, but it wouldn't have +# moto's custom service backends +SERVICE_NAMES = Union[ + "Literal['acm']", + "Literal['acm-pca']", + "Literal['amp']", + "Literal['apigateway']", + "Literal['apigatewaymanagementapi']", + "Literal['apigatewayv2']", + "Literal['appconfig']", + "Literal['applicationautoscaling']", + "Literal['appsync']", + "Literal['athena']", + "Literal['autoscaling']", + "Literal['batch']", + "Literal['budgets']", + "Literal['ce']", + "Literal['cloudformation']", + "Literal['cloudfront']", + "Literal['cloudtrail']", + "Literal['cloudwatch']", + "Literal['codebuild']", + "Literal['codecommit']", + "Literal['codepipeline']", + "Literal['cognito-identity']", + "Literal['cognito-idp']", + "Literal['comprehend']", + "Literal['config']", + "Literal['databrew']", + "Literal['datapipeline']", + "Literal['datasync']", + "Literal['dax']", + "Literal['dms']", + "Literal['ds']", + "Literal['dynamodb']", + "Literal['dynamodb_v20111205']", + "Literal['dynamodbstreams']", + "Literal['ebs']", + "Literal['ec2']", + "Literal['ec2instanceconnect']", + "Literal['ecr']", + "Literal['ecs']", + "Literal['efs']", + "Literal['eks']", + "Literal['elasticache']", + "Literal['elasticbeanstalk']", + "Literal['elastictranscoder']", + "Literal['elb']", + "Literal['elbv2']", + "Literal['emr']", + "Literal['emr-containers']", + "Literal['emr-serverless']", + "Literal['es']", + "Literal['events']", + "Literal['firehose']", + "Literal['forecast']", + "Literal['glacier']", + "Literal['glue']", + "Literal['greengrass']", + "Literal['guardduty']", + "Literal['iam']", + "Literal['identitystore']", + "Literal['inspector2']", + "Literal['instance_metadata']", + "Literal['iot']", + "Literal['iot-data']", + "Literal['ivs']", + "Literal['kinesis']", + "Literal['kinesisvideo']", + "Literal['kinesis-video-archived-media']", + "Literal['kms']", + "Literal['lakeformation']", + "Literal['lambda']", + "Literal['logs']", + "Literal['managedblockchain']", + "Literal['mediaconnect']", + "Literal['medialive']", + "Literal['mediapackage']", + "Literal['mediastore']", + "Literal['mediastore-data']", + "Literal['meteringmarketplace']", + "Literal['moto_api']", + "Literal['mq']", + "Literal['neptune']", + "Literal['opensearch']", + "Literal['opsworks']", + "Literal['organizations']", + "Literal['personalize']", + "Literal['pinpoint']", + "Literal['polly']", + "Literal['quicksight']", + "Literal['ram']", + "Literal['rds']", + "Literal['rds-data']", + "Literal['redshift']", + "Literal['redshift-data']", + "Literal['rekognition']", + "Literal['resource-groups']", + "Literal['resourcegroupstaggingapi']", + "Literal['robomaker']", + "Literal['route53']", + "Literal['route53resolver']", + "Literal['s3']", + "Literal['s3bucket_path']", + "Literal['s3control']", + "Literal['sagemaker']", + "Literal['sagemaker-runtime']", + "Literal['scheduler']", + "Literal['sdb']", + "Literal['secretsmanager']", + "Literal['servicediscovery']", + "Literal['service-quotas']", + "Literal['ses']", + "Literal['sesv2']", + "Literal['signer']", + "Literal['sns']", + "Literal['sqs']", + "Literal['ssm']", + "Literal['sso-admin']", + "Literal['stepfunctions']", + "Literal['sts']", + "Literal['support']", + "Literal['swf']", + "Literal['textract']", + "Literal['timestream-write']", + "Literal['transcribe']", + "Literal['wafv2']", + "Literal['xray']", +] -def _import_backend(module_name: str, backends_name: str) -> BackendDict: + +def _import_backend( + module_name: str, + backends_name: str, +) -> "BackendDict[SERVICE_BACKEND]": module = importlib.import_module("moto." + module_name) return getattr(module, backends_name) -def backends() -> Iterable[BackendDict]: +def backends() -> "Iterable[BackendDict[BaseBackend]]": for module_name, backends_name in BACKENDS.values(): yield _import_backend(module_name, backends_name) -def service_backends() -> Iterable[BackendDict]: +def service_backends() -> "Iterable[BackendDict[BaseBackend]]": services = [(f.name, f.backend) for f in decorator_functions] for module_name, backends_name in sorted(set(services)): yield _import_backend(module_name, backends_name) -def loaded_backends() -> Iterable[Tuple[str, BackendDict]]: +def loaded_backends() -> "Iterable[Tuple[str, BackendDict[BaseBackend]]]": loaded_modules = sys.modules.keys() moto_modules = [m for m in loaded_modules if m.startswith("moto.")] imported_backends = [ @@ -43,6 +307,263 @@ def loaded_backends() -> Iterable[Tuple[str, BackendDict]]: yield name, _import_backend(module_name, backends_name) -def get_backend(name: str) -> BackendDict: +# fmt: off +# This is more or less the dummy-implementation style that's currently in black's preview +# style. It should be live in black v24.0+, at which point we should remove the # fmt: off +# directive. +@overload +def get_backend(name: "Literal['acm']") -> "BackendDict[AWSCertificateManagerBackend]": ... +@overload +def get_backend(name: "Literal['acm-pca']") -> "BackendDict[ACMPCABackend]": ... +@overload +def get_backend(name: "Literal['amp']") -> "BackendDict[PrometheusServiceBackend]": ... +@overload +def get_backend(name: "Literal['apigateway']") -> "BackendDict[APIGatewayBackend]": ... +@overload +def get_backend(name: "Literal['apigatewaymanagementapi']") -> "BackendDict[ApiGatewayManagementApiBackend]": ... +@overload +def get_backend(name: "Literal['apigatewayv2']") -> "BackendDict[ApiGatewayV2Backend]": ... +@overload +def get_backend(name: "Literal['appconfig']") -> "BackendDict[AppConfigBackend]": ... +@overload +def get_backend(name: "Literal['applicationautoscaling']") -> "BackendDict[ApplicationAutoscalingBackend]": ... +@overload +def get_backend(name: "Literal['appsync']") -> "BackendDict[AppSyncBackend]": ... +@overload +def get_backend(name: "Literal['athena']") -> "BackendDict[AthenaBackend]": ... +@overload +def get_backend(name: "Literal['autoscaling']") -> "BackendDict[AutoScalingBackend]": ... +@overload +def get_backend(name: "Literal['batch']") -> "BackendDict[BatchBackend]": ... +@overload +def get_backend(name: "Literal['budgets']") -> "BackendDict[BudgetsBackend]": ... +@overload +def get_backend(name: "Literal['ce']") -> "BackendDict[CostExplorerBackend]": ... +@overload +def get_backend(name: "Literal['cloudformation']") -> "BackendDict[CloudFormationBackend]": ... +@overload +def get_backend(name: "Literal['cloudfront']") -> "BackendDict[CloudFrontBackend]": ... +@overload +def get_backend(name: "Literal['cloudtrail']") -> "BackendDict[CloudTrailBackend]": ... +@overload +def get_backend(name: "Literal['cloudwatch']") -> "BackendDict[CloudWatchBackend]": ... +@overload +def get_backend(name: "Literal['codebuild']") -> "BackendDict[CodeBuildBackend]": ... +@overload +def get_backend(name: "Literal['codecommit']") -> "BackendDict[CodeCommitBackend]": ... +@overload +def get_backend(name: "Literal['codepipeline']") -> "BackendDict[CodePipelineBackend]": ... +@overload +def get_backend(name: "Literal['cognito-identity']") -> "BackendDict[CognitoIdentityBackend]": ... +@overload +def get_backend(name: "Literal['cognito-idp']") -> "BackendDict[CognitoIdpBackend]": ... +@overload +def get_backend(name: "Literal['comprehend']") -> "BackendDict[ComprehendBackend]": ... +@overload +def get_backend(name: "Literal['config']") -> "BackendDict[ConfigBackend]": ... +@overload +def get_backend(name: "Literal['databrew']") -> "BackendDict[DataBrewBackend]": ... +@overload +def get_backend(name: "Literal['datapipeline']") -> "BackendDict[DataPipelineBackend]": ... +@overload +def get_backend(name: "Literal['datasync']") -> "BackendDict[DataSyncBackend]": ... +@overload +def get_backend(name: "Literal['dax']") -> "BackendDict[DAXBackend]": ... +@overload +def get_backend(name: "Literal['dms']") -> "BackendDict[DatabaseMigrationServiceBackend]": ... +@overload +def get_backend(name: "Literal['ds']") -> "BackendDict[DirectoryServiceBackend]": ... +@overload +def get_backend(name: "Literal['dynamodb']") -> "BackendDict[DynamoDBBackend]": ... +@overload +def get_backend(name: "Literal['dynamodb_v20111205']") -> "BackendDict[DynamoDBBackend_v20111205]": ... +@overload +def get_backend(name: "Literal['dynamodbstreams']") -> "BackendDict[DynamoDBStreamsBackend]": ... +@overload +def get_backend(name: "Literal['ebs']") -> "BackendDict[EBSBackend]": ... +@overload +def get_backend(name: "Literal['ec2']") -> "BackendDict[EC2Backend]": ... +@overload +def get_backend(name: "Literal['ec2instanceconnect']") -> "BackendDict[Ec2InstanceConnectBackend]": ... +@overload +def get_backend(name: "Literal['ecr']") -> "BackendDict[ECRBackend]": ... +@overload +def get_backend(name: "Literal['ecs']") -> "BackendDict[EC2ContainerServiceBackend]": ... +@overload +def get_backend(name: "Literal['efs']") -> "BackendDict[EFSBackend]": ... +@overload +def get_backend(name: "Literal['eks']") -> "BackendDict[EKSBackend]": ... +@overload +def get_backend(name: "Literal['elasticache']") -> "BackendDict[ElastiCacheBackend]": ... +@overload +def get_backend(name: "Literal['elasticbeanstalk']") -> "BackendDict[EBBackend]": ... +@overload +def get_backend(name: "Literal['elastictranscoder']") -> "BackendDict[ElasticTranscoderBackend]": ... +@overload +def get_backend(name: "Literal['elb']") -> "BackendDict[ELBBackend]": ... +@overload +def get_backend(name: "Literal['elbv2']") -> "BackendDict[ELBv2Backend]": ... +@overload +def get_backend(name: "Literal['emr']") -> "BackendDict[ElasticMapReduceBackend]": ... +@overload +def get_backend(name: "Literal['emr-containers']") -> "BackendDict[EMRContainersBackend]": ... +@overload +def get_backend(name: "Literal['emr-serverless']") -> "BackendDict[EMRServerlessBackend]": ... +@overload +def get_backend(name: "Literal['es']") -> "BackendDict[ElasticsearchServiceBackend]": ... +@overload +def get_backend(name: "Literal['events']") -> "BackendDict[EventsBackend]": ... +@overload +def get_backend(name: "Literal['firehose']") -> "BackendDict[FirehoseBackend]": ... +@overload +def get_backend(name: "Literal['forecast']") -> "BackendDict[ForecastBackend]": ... +@overload +def get_backend(name: "Literal['glacier']") -> "BackendDict[GlacierBackend]": ... +@overload +def get_backend(name: "Literal['glue']") -> "BackendDict[GlueBackend]": ... +@overload +def get_backend(name: "Literal['greengrass']") -> "BackendDict[GreengrassBackend]": ... +@overload +def get_backend(name: "Literal['guardduty']") -> "BackendDict[GuardDutyBackend]": ... +@overload +def get_backend(name: "Literal['iam']") -> "BackendDict[IAMBackend]": ... +@overload +def get_backend(name: "Literal['identitystore']") -> "BackendDict[IdentityStoreBackend]": ... +@overload +def get_backend(name: "Literal['inspector2']") -> "BackendDict[Inspector2Backend]": ... +@overload +def get_backend(name: "Literal['instance_metadata']") -> "BackendDict[InstanceMetadataBackend]": ... +@overload +def get_backend(name: "Literal['iot']") -> "BackendDict[IoTBackend]": ... +@overload +def get_backend(name: "Literal['iot-data']") -> "BackendDict[IoTDataPlaneBackend]": ... +@overload +def get_backend(name: "Literal['ivs']") -> "BackendDict[IVSBackend]": ... +@overload +def get_backend(name: "Literal['kinesis']") -> "BackendDict[KinesisBackend]": ... +@overload +def get_backend(name: "Literal['kinesisvideo']") -> "BackendDict[KinesisVideoBackend]": ... +@overload +def get_backend(name: "Literal['kinesis-video-archived-media']") -> "BackendDict[KinesisVideoArchivedMediaBackend]": ... +@overload +def get_backend(name: "Literal['kms']") -> "BackendDict[KmsBackend]": ... +@overload +def get_backend(name: "Literal['lakeformation']") -> "BackendDict[LakeFormationBackend]": ... +@overload +def get_backend(name: "Literal['lambda']") -> "BackendDict[LambdaBackend]": ... +@overload +def get_backend(name: "Literal['logs']") -> "BackendDict[LogsBackend]": ... +@overload +def get_backend(name: "Literal['managedblockchain']") -> "BackendDict[ManagedBlockchainBackend]": ... +@overload +def get_backend(name: "Literal['mediaconnect']") -> "BackendDict[MediaConnectBackend]": ... +@overload +def get_backend(name: "Literal['medialive']") -> "BackendDict[MediaLiveBackend]": ... +@overload +def get_backend(name: "Literal['mediapackage']") -> "BackendDict[MediaPackageBackend]": ... +@overload +def get_backend(name: "Literal['mediastore']") -> "BackendDict[MediaStoreBackend]": ... +@overload +def get_backend(name: "Literal['mediastore-data']") -> "BackendDict[MediaStoreDataBackend]": ... +@overload +def get_backend(name: "Literal['meteringmarketplace']") -> "BackendDict[MeteringMarketplaceBackend]": ... +@overload +def get_backend(name: "Literal['moto_api']") -> "BackendDict[MotoAPIBackend]": ... +@overload +def get_backend(name: "Literal['mq']") -> "BackendDict[MQBackend]": ... +@overload +def get_backend(name: "Literal['neptune']") -> "BackendDict[NeptuneBackend]": ... +@overload +def get_backend(name: "Literal['opensearch']") -> "BackendDict[OpenSearchServiceBackend]": ... +@overload +def get_backend(name: "Literal['opsworks']") -> "BackendDict[OpsWorksBackend]": ... +@overload +def get_backend(name: "Literal['organizations']") -> "BackendDict[OrganizationsBackend]": ... +@overload +def get_backend(name: "Literal['personalize']") -> "BackendDict[PersonalizeBackend]": ... +@overload +def get_backend(name: "Literal['pinpoint']") -> "BackendDict[PinpointBackend]": ... +@overload +def get_backend(name: "Literal['polly']") -> "BackendDict[PollyBackend]": ... +@overload +def get_backend(name: "Literal['quicksight']") -> "BackendDict[QuickSightBackend]": ... +@overload +def get_backend(name: "Literal['ram']") -> "BackendDict[ResourceAccessManagerBackend]": ... +@overload +def get_backend(name: "Literal['rds']") -> "BackendDict[RDSBackend]": ... +@overload +def get_backend(name: "Literal['rds-data']") -> "BackendDict[RDSDataServiceBackend]": ... +@overload +def get_backend(name: "Literal['redshift']") -> "BackendDict[RedshiftBackend]": ... +@overload +def get_backend(name: "Literal['redshift-data']") -> "BackendDict[RedshiftDataAPIServiceBackend]": ... +@overload +def get_backend(name: "Literal['rekognition']") -> "BackendDict[RekognitionBackend]": ... +@overload +def get_backend(name: "Literal['resource-groups']") -> "BackendDict[ResourceGroupsBackend]": ... +@overload +def get_backend(name: "Literal['resourcegroupstaggingapi']") -> "BackendDict[ResourceGroupsTaggingAPIBackend]": ... +@overload +def get_backend(name: "Literal['robomaker']") -> "BackendDict[RoboMakerBackend]": ... +@overload +def get_backend(name: "Literal['route53']") -> "BackendDict[Route53Backend]": ... +@overload +def get_backend(name: "Literal['route53resolver']") -> "BackendDict[Route53ResolverBackend]": ... +@overload +def get_backend(name: "Literal['s3']") -> "BackendDict[S3Backend]": ... +@overload +def get_backend(name: "Literal['s3bucket_path']") -> "BackendDict[S3Backend]": ... +@overload +def get_backend(name: "Literal['s3control']") -> "BackendDict[S3ControlBackend]": ... +@overload +def get_backend(name: "Literal['sagemaker']") -> "BackendDict[SageMakerModelBackend]": ... +@overload +def get_backend(name: "Literal['sagemaker-runtime']") -> "BackendDict[SageMakerRuntimeBackend]": ... +@overload +def get_backend(name: "Literal['scheduler']") -> "BackendDict[EventBridgeSchedulerBackend]": ... +@overload +def get_backend(name: "Literal['sdb']") -> "BackendDict[SimpleDBBackend]": ... +@overload +def get_backend(name: "Literal['secretsmanager']") -> "BackendDict[SecretsManagerBackend]": ... +@overload +def get_backend(name: "Literal['servicediscovery']") -> "BackendDict[ServiceDiscoveryBackend]": ... +@overload +def get_backend(name: "Literal['service-quotas']") -> "BackendDict[ServiceQuotasBackend]": ... +@overload +def get_backend(name: "Literal['ses']") -> "BackendDict[SESBackend]": ... +@overload +def get_backend(name: "Literal['sesv2']") -> "BackendDict[SESV2Backend]": ... +@overload +def get_backend(name: "Literal['signer']") -> "BackendDict[SignerBackend]": ... +@overload +def get_backend(name: "Literal['sns']") -> "BackendDict[SNSBackend]": ... +@overload +def get_backend(name: "Literal['sqs']") -> "BackendDict[SQSBackend]": ... +@overload +def get_backend(name: "Literal['ssm']") -> "BackendDict[SimpleSystemManagerBackend]": ... +@overload +def get_backend(name: "Literal['sso-admin']") -> "BackendDict[SSOAdminBackend]": ... +@overload +def get_backend(name: "Literal['stepfunctions']") -> "BackendDict[StepFunctionBackend]": ... +@overload +def get_backend(name: "Literal['sts']") -> "BackendDict[STSBackend]": ... +@overload +def get_backend(name: "Literal['support']") -> "BackendDict[SupportBackend]": ... +@overload +def get_backend(name: "Literal['swf']") -> "BackendDict[SWFBackend]": ... +@overload +def get_backend(name: "Literal['textract']") -> "BackendDict[TextractBackend]": ... +@overload +def get_backend(name: "Literal['timestream-write']") -> "BackendDict[TimestreamWriteBackend]": ... +@overload +def get_backend(name: "Literal['transcribe']") -> "BackendDict[TranscribeBackend]": ... +@overload +def get_backend(name: "Literal['wafv2']") -> "BackendDict[WAFV2Backend]": ... +@overload +def get_backend(name: "Literal['xray']") -> "BackendDict[XRayBackend]": ... +# fmt: on + + +def get_backend(name: SERVICE_NAMES) -> "BackendDict[SERVICE_BACKEND]": module_name, backends_name = BACKENDS[name] return _import_backend(module_name, backends_name) diff --git a/moto/batch/models.py b/moto/batch/models.py index a9768c107..0f6879ae9 100644 --- a/moto/batch/models.py +++ b/moto/batch/models.py @@ -211,7 +211,7 @@ class JobQueue(CloudFormationModel): priority=properties["Priority"], state=properties.get("State", "ENABLED"), compute_env_order=compute_envs, - schedule_policy={}, + schedule_policy=None, ) @@ -448,18 +448,18 @@ class JobDefinition(CloudFormationModel): tags=lowercase_first_key(properties.get("Tags", {})), retry_strategy=lowercase_first_key(properties["RetryStrategy"]), container_properties=( - lowercase_first_key(properties["ContainerProperties"]) + lowercase_first_key(properties["ContainerProperties"]) # type: ignore[arg-type] if "ContainerProperties" in properties else None ), node_properties=( - lowercase_first_key(properties["NodeProperties"]) + lowercase_first_key(properties["NodeProperties"]) # type: ignore[arg-type] if "NodeProperties" in properties else None ), timeout=lowercase_first_key(properties.get("timeout", {})), - platform_capabilities=None, - propagate_tags=None, + platform_capabilities=None, # type: ignore[arg-type] + propagate_tags=None, # type: ignore[arg-type] ) diff --git a/moto/batch_simple/responses.py b/moto/batch_simple/responses.py index dfb9e75e9..32c336e32 100644 --- a/moto/batch_simple/responses.py +++ b/moto/batch_simple/responses.py @@ -9,4 +9,4 @@ class BatchSimpleResponse(BatchResponse): :return: Batch Backend :rtype: moto.batch.models.BatchBackend """ - return batch_simple_backends[self.current_account][self.region] + return batch_simple_backends[self.current_account][self.region] # type: ignore[return-value] diff --git a/moto/cloudformation/parsing.py b/moto/cloudformation/parsing.py index a570b043f..a798a33ef 100644 --- a/moto/cloudformation/parsing.py +++ b/moto/cloudformation/parsing.py @@ -598,9 +598,10 @@ class ResourceMap(collections_abc.Mapping): # type: ignore[type-arg] location = params["Location"] bucket_name, name = bucket_and_name_from_url(location) key = s3_backends[self._account_id]["global"].get_object( - bucket_name, name + bucket_name, # type: ignore[arg-type] + name, ) - self._parsed_resources.update(json.loads(key.value)) + self._parsed_resources.update(json.loads(key.value)) # type: ignore[union-attr] def parse_ssm_parameter(self, value: str, value_type: str) -> str: # The Value in SSM parameters is the SSM parameter path @@ -609,9 +610,9 @@ class ResourceMap(collections_abc.Mapping): # type: ignore[type-arg] parameter = ssm_backends[self._account_id][self._region_name].get_parameter( value ) - actual_value = parameter.value + actual_value = parameter.value # type: ignore[union-attr] if value_type.find("List") > 0: - return actual_value.split(",") + return actual_value.split(",") # type: ignore[return-value] return actual_value def load_parameters(self) -> None: diff --git a/moto/cloudformation/utils.py b/moto/cloudformation/utils.py index 965db2b2e..7225b5c21 100644 --- a/moto/cloudformation/utils.py +++ b/moto/cloudformation/utils.py @@ -109,4 +109,4 @@ def get_stack_from_s3_url(template_url: str, account_id: str) -> str: key_name = template_url_parts.path.lstrip("/") key = s3_backends[account_id]["global"].get_object(bucket_name, key_name) - return key.value.decode("utf-8") + return key.value.decode("utf-8") # type: ignore[union-attr] diff --git a/moto/cognitoidp/responses.py b/moto/cognitoidp/responses.py index a302ba881..dbe3ae509 100644 --- a/moto/cognitoidp/responses.py +++ b/moto/cognitoidp/responses.py @@ -469,7 +469,7 @@ class CognitoIdpResponse(BaseResponse): ].forgot_password(client_id, username) self.response_headers[ "x-moto-forgot-password-confirmation-code" - ] = confirmation_code + ] = confirmation_code # type: ignore[assignment] return json.dumps(response) # This endpoint receives no authorization header, so if moto-server is listening diff --git a/moto/core/__init__.py b/moto/core/__init__.py index 140e82694..b202845dc 100644 --- a/moto/core/__init__.py +++ b/moto/core/__init__.py @@ -1,5 +1,5 @@ from .models import DEFAULT_ACCOUNT_ID # noqa -from .base_backend import BaseBackend, BackendDict # noqa +from .base_backend import BaseBackend, BackendDict, SERVICE_BACKEND # noqa from .common_models import BaseModel # noqa from .common_models import CloudFormationModel, CloudWatchMetricProvider # noqa from .models import patch_client, patch_resource # noqa diff --git a/moto/core/base_backend.py b/moto/core/base_backend.py index d05a46526..92fba97a1 100644 --- a/moto/core/base_backend.py +++ b/moto/core/base_backend.py @@ -2,7 +2,7 @@ import re import string from functools import lru_cache from threading import RLock -from typing import Any, ClassVar, Dict, Iterator, List, Optional, TypeVar +from typing import Any, Callable, ClassVar, Dict, Iterator, List, Optional, TypeVar from uuid import uuid4 from boto3 import Session @@ -10,6 +10,7 @@ from boto3 import Session from moto.settings import allow_unknown_region, enable_iso_regions from .model_instances import model_data +from .responses import TYPE_RESPONSE from .utils import convert_regex_to_flask_path @@ -47,7 +48,7 @@ class BaseBackend: return backend_urls_module @property - def urls(self) -> Dict[str, str]: + def urls(self) -> Dict[str, Callable[[Any, str, Any], TYPE_RESPONSE]]: # type: ignore[misc] """ A dictionary of the urls to be mocked with this service and the handlers that should be called in their place @@ -81,7 +82,7 @@ class BaseBackend: return urls @property - def url_paths(self) -> Dict[str, str]: + def url_paths(self) -> Dict[str, Callable[[Any, str, Any], TYPE_RESPONSE]]: # type: ignore[misc] """ A dictionary of the paths of the urls to be mocked with this service and the handlers that should be called in their place @@ -103,7 +104,7 @@ class BaseBackend: return self._url_module.url_bases @property - def flask_paths(self) -> Dict[str, str]: + def flask_paths(self) -> Dict[str, Callable[[Any, str, Any], TYPE_RESPONSE]]: # type: ignore[misc] """ The url paths that will be used for the flask server """ @@ -259,7 +260,7 @@ class AccountSpecificBackend(Dict[str, SERVICE_BACKEND]): return super().__getitem__(region_name) -class BackendDict(Dict[str, AccountSpecificBackend]): # type: ignore[type-arg] +class BackendDict(Dict[str, AccountSpecificBackend[SERVICE_BACKEND]]): """ Data Structure to store everything related to a specific service. Format: @@ -269,7 +270,7 @@ class BackendDict(Dict[str, AccountSpecificBackend]): # type: ignore[type-arg] def __init__( self, - backend: Any, + backend: "type[SERVICE_BACKEND]", service_name: str, use_boto3_regions: bool = True, additional_regions: Optional[List[str]] = None, @@ -292,7 +293,7 @@ class BackendDict(Dict[str, AccountSpecificBackend]): # type: ignore[type-arg] return not self.__eq__(other) @lru_cache() - def __getitem__(self, account_id: str) -> AccountSpecificBackend: # type: ignore + def __getitem__(self, account_id: str) -> AccountSpecificBackend[SERVICE_BACKEND]: self._create_account_specific_backend(account_id) return super().__getitem__(account_id) @@ -305,7 +306,11 @@ class BackendDict(Dict[str, AccountSpecificBackend]): # type: ignore[type-arg] def __len__(self) -> int: return super().__len__() - def __setitem__(self, key: str, value: AccountSpecificBackend) -> None: # type: ignore[type-arg] + def __setitem__( + self, + key: str, + value: AccountSpecificBackend[SERVICE_BACKEND], + ) -> None: super().__setitem__(key, value) def _create_account_specific_backend(self, account_id: str) -> None: diff --git a/moto/core/models.py b/moto/core/models.py index 1bcada3e7..8017c8635 100644 --- a/moto/core/models.py +++ b/moto/core/models.py @@ -27,7 +27,7 @@ from botocore.handlers import BUILTIN_HANDLERS from moto import settings -from .base_backend import BackendDict +from .base_backend import SERVICE_BACKEND, BackendDict, BaseBackend from .botocore_stubber import BotocoreStubber from .custom_responses_mock import ( CallbackResponse, @@ -53,13 +53,13 @@ class BaseMockAWS(ContextManager["BaseMockAWS"]): nested_count = 0 mocks_active = False - def __init__(self, backends: BackendDict): + def __init__(self, backends: BackendDict[SERVICE_BACKEND]): from moto.instance_metadata import instance_metadata_backends from moto.moto_api._internal.models import moto_api_backend self.backends = backends - self.backends_for_urls = [] + self.backends_for_urls: list[BaseBackend] = [] default_account_id = DEFAULT_ACCOUNT_ID default_backends = [ instance_metadata_backends[default_account_id]["global"], @@ -534,7 +534,7 @@ class ProxyModeMockAWS(BaseMockAWS): class base_decorator: mock_backend = MockAWS - def __init__(self, backends: BackendDict): + def __init__(self, backends: BackendDict[SERVICE_BACKEND]): self.backends = backends @overload diff --git a/moto/ds/models.py b/moto/ds/models.py index 679cb33be..1842643b7 100644 --- a/moto/ds/models.py +++ b/moto/ds/models.py @@ -133,13 +133,13 @@ class Directory(BaseModel): # pylint: disable=too-many-instance-attributes self.region ].create_network_interface( subnet=subnet_id, - private_ip_address=None, + private_ip_address=None, # type: ignore[arg-type] group_ids=[security_group_id], description=f"AWS created network interface for {self.directory_id}", ) eni_ids.append(eni_info.id) subnet_ips.append(eni_info.private_ip_address) - return eni_ids, subnet_ips + return eni_ids, subnet_ips # type: ignore[return-value] def delete_eni(self) -> None: """Delete ENI for each subnet and the security group.""" diff --git a/moto/ec2/models/__init__.py b/moto/ec2/models/__init__.py index 97135e991..bd448918a 100644 --- a/moto/ec2/models/__init__.py +++ b/moto/ec2/models/__init__.py @@ -73,7 +73,7 @@ class SettingsBackend: ec2_backend = ec2_backends[self.account_id][self.region_name] # type: ignore[attr-defined] ec2_backend.ebs_encryption_by_default = True - def get_ebs_encryption_by_default(self) -> None: + def get_ebs_encryption_by_default(self) -> bool: ec2_backend = ec2_backends[self.account_id][self.region_name] # type: ignore[attr-defined] return ec2_backend.ebs_encryption_by_default diff --git a/moto/ec2/models/elastic_block_store.py b/moto/ec2/models/elastic_block_store.py index f2ccb515a..ac094d757 100644 --- a/moto/ec2/models/elastic_block_store.py +++ b/moto/ec2/models/elastic_block_store.py @@ -96,12 +96,11 @@ class VolumeAttachment(CloudFormationModel): volume_id = properties["VolumeId"] ec2_backend = ec2_backends[account_id][region_name] - attachment = ec2_backend.attach_volume( + return ec2_backend.attach_volume( # type: ignore[return-value] volume_id=volume_id, instance_id=instance_id, device_path=properties["Device"], ) - return attachment class Volume(TaggedEC2Resource, CloudFormationModel): diff --git a/moto/ec2/models/instances.py b/moto/ec2/models/instances.py index 1c94338cc..f76ebce45 100644 --- a/moto/ec2/models/instances.py +++ b/moto/ec2/models/instances.py @@ -296,7 +296,7 @@ class Instance(TaggedEC2Resource, BotoInstance, CloudFormationModel): ec2_backend = ec2_backends[account_id][region_name] security_group_ids = properties.get("SecurityGroups", []) group_names = [ - ec2_backend.get_security_group_from_id(group_id).name + ec2_backend.get_security_group_from_id(group_id).name # type: ignore[union-attr] for group_id in security_group_ids ] diff --git a/moto/ec2/models/subnets.py b/moto/ec2/models/subnets.py index faece0061..1d99d484f 100644 --- a/moto/ec2/models/subnets.py +++ b/moto/ec2/models/subnets.py @@ -1,12 +1,13 @@ import ipaddress import itertools from collections import defaultdict -from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Set, Tuple +from typing import TYPE_CHECKING, Any, Dict, List, Optional, Set, Tuple from moto.core import CloudFormationModel if TYPE_CHECKING: from moto.ec2.models.instances import Instance + from moto.ec2.models.availability_zones_and_regions import Zone from ..exceptions import ( @@ -412,7 +413,7 @@ class SubnetBackend: def describe_subnets( self, subnet_ids: Optional[List[str]] = None, filters: Optional[Any] = None - ) -> Iterable[Subnet]: + ) -> List[Subnet]: # Extract a list of all subnets matches = list( itertools.chain(*[x.copy().values() for x in self.subnets.copy().values()]) diff --git a/moto/elbv2/models.py b/moto/elbv2/models.py index fad376219..4b3cf67ea 100644 --- a/moto/elbv2/models.py +++ b/moto/elbv2/models.py @@ -421,7 +421,7 @@ class FakeListenerRule(CloudFormationModel): listener_rule = elbv2_backend.modify_rule( original_resource.arn, conditions, actions ) - return listener_rule + return listener_rule # type: ignore[return-value] class FakeRule(BaseModel): diff --git a/moto/events/models.py b/moto/events/models.py index f413eca8c..386168774 100644 --- a/moto/events/models.py +++ b/moto/events/models.py @@ -187,8 +187,8 @@ class Rule(CloudFormationModel): archive = events_backends[self.account_id][self.region_name].archives.get( archive_name ) - if archive.uuid == archive_uuid: - archive.events.append(event) + if archive.uuid == archive_uuid: # type: ignore[union-attr] + archive.events.append(event) # type: ignore[arg-type,union-attr] def _find_api_destination(self, resource_id: str) -> "Destination": backend: "EventsBackend" = events_backends[self.account_id][self.region_name] diff --git a/moto/events/notifications.py b/moto/events/notifications.py index 78da67f13..67e13049d 100644 --- a/moto/events/notifications.py +++ b/moto/events/notifications.py @@ -53,7 +53,7 @@ def _send_safe_notification( for target in applicable_targets: if target.get("Arn", "").startswith("arn:aws:lambda"): - _invoke_lambda(account_id, target.get("Arn"), event=event) + _invoke_lambda(account_id, target.get("Arn"), event=event) # type: ignore[arg-type] def _invoke_lambda(account_id: str, fn_arn: str, event: Any) -> None: @@ -64,7 +64,7 @@ def _invoke_lambda(account_id: str, fn_arn: str, event: Any) -> None: body = json.dumps(event) lambda_backends[account_id][lmbda_region].invoke( function_name=fn_arn, - qualifier=None, + qualifier=None, # type: ignore[arg-type] body=body, headers=dict(), response_headers=dict(), diff --git a/moto/iam/models.py b/moto/iam/models.py index 6438b7f72..f555874e9 100644 --- a/moto/iam/models.py +++ b/moto/iam/models.py @@ -675,7 +675,7 @@ class Role(CloudFormationModel): self.tags = tags # last_used should be treated as part of the public API # https://github.com/getmoto/moto/issues/6859 - self.last_used = None + self.last_used: Optional[datetime] = None self.last_used_region = None self.description = description self.permissions_boundary: Optional[str] = permissions_boundary diff --git a/moto/iot/models.py b/moto/iot/models.py index 4846ad5f5..359dd5a83 100644 --- a/moto/iot/models.py +++ b/moto/iot/models.py @@ -1255,7 +1255,7 @@ class IoTBackend(BaseBackend): cognito = cognitoidentity_backends[self.account_id][self.region_name] identities = [] for identity_pool in cognito.identity_pools: - pool_identities = cognito.pools_identities.get(identity_pool, []) + pool_identities = cognito.pools_identities.get(identity_pool, {}) identities.extend( [pi["IdentityId"] for pi in pool_identities.get("Identities", [])] ) diff --git a/moto/logs/models.py b/moto/logs/models.py index 08d1780b4..e7ac94a36 100644 --- a/moto/logs/models.py +++ b/moto/logs/models.py @@ -1266,7 +1266,7 @@ class LogsBackend(BaseBackend): logGroupName, [], fromTime, to, None, None, "", False ) s3_backends[self.account_id]["global"].put_object( - destination, destinationPrefix, json.dumps(logs) + destination, destinationPrefix, json.dumps(logs).encode("utf-8") ) self.export_tasks[task_id].status["code"] = "completed" self.export_tasks[task_id].status["message"] = "Task is completed" diff --git a/moto/moto_server/werkzeug_app.py b/moto/moto_server/werkzeug_app.py index 5fde1d409..25b76ad17 100644 --- a/moto/moto_server/werkzeug_app.py +++ b/moto/moto_server/werkzeug_app.py @@ -57,7 +57,9 @@ class DomainDispatcherApplication: """ def __init__( - self, create_app: Callable[[str], Flask], service: Optional[str] = None + self, + create_app: Callable[[backends.SERVICE_NAMES], Flask], + service: Optional[str] = None, ): self.create_app = create_app self.lock = Lock() @@ -260,7 +262,7 @@ class DomainDispatcherApplication: return backend_app(environ, start_response) -def create_backend_app(service: str) -> Flask: +def create_backend_app(service: backends.SERVICE_NAMES) -> Flask: from werkzeug.routing import Map current_file = os.path.abspath(__file__) @@ -292,7 +294,7 @@ def create_backend_app(service: str) -> Flask: for url_path, handler in backend.flask_paths.items(): view_func = convert_to_flask_response(handler) if handler.__name__ == "dispatch": - endpoint = f"{handler.__self__.__name__}.dispatch" + endpoint = f"{handler.__self__.__name__}.dispatch" # type: ignore[attr-defined] else: endpoint = view_func.__name__ diff --git a/moto/rds/models.py b/moto/rds/models.py index f7dd2b680..250330b5c 100644 --- a/moto/rds/models.py +++ b/moto/rds/models.py @@ -4,7 +4,7 @@ import re import string from collections import OrderedDict, defaultdict from re import compile as re_compile -from typing import Any, Dict, Iterable, List, Optional, Tuple, Union +from typing import TYPE_CHECKING, Any, Dict, Iterable, List, Optional, Tuple, Union from jinja2 import Template @@ -54,6 +54,9 @@ from .utils import ( validate_filters, ) +if TYPE_CHECKING: + from moto.ec2.models.subnets import Subnet + def find_cluster(cluster_arn: str) -> "Cluster": arn_parts = cluster_arn.split(":") @@ -620,12 +623,11 @@ class Database(CloudFormationModel): self.availability_zone = f"{self.region_name}a" self.multi_az = kwargs.get("multi_az") self.db_subnet_group_name = kwargs.get("db_subnet_group_name") + self.db_subnet_group = None if self.db_subnet_group_name: self.db_subnet_group = rds_backends[self.account_id][ self.region_name ].describe_db_subnet_groups(self.db_subnet_group_name)[0] - else: - self.db_subnet_group = None self.security_groups = kwargs.get("security_groups", []) self.vpc_security_group_ids = kwargs.get("vpc_security_group_ids", []) self.preferred_maintenance_window = kwargs.get("preferred_maintenance_window") @@ -1092,7 +1094,7 @@ class Database(CloudFormationModel): def delete(self, account_id: str, region_name: str) -> None: backend = rds_backends[account_id][region_name] - backend.delete_db_instance(self.db_instance_identifier) + backend.delete_db_instance(self.db_instance_identifier) # type: ignore[arg-type] class DatabaseSnapshot(BaseModel): @@ -1403,10 +1405,10 @@ class SecurityGroup(CloudFormationModel): security_group.authorize_cidr(ingress_value) elif ingress_type == "EC2SecurityGroupName": subnet = ec2_backend.get_security_group_from_name(ingress_value) - security_group.authorize_security_group(subnet) + security_group.authorize_security_group(subnet) # type: ignore[arg-type] elif ingress_type == "EC2SecurityGroupId": subnet = ec2_backend.get_security_group_from_id(ingress_value) - security_group.authorize_security_group(subnet) + security_group.authorize_security_group(subnet) # type: ignore[arg-type] return security_group def get_tags(self) -> List[Dict[str, str]]: @@ -1431,7 +1433,7 @@ class SubnetGroup(CloudFormationModel): self, subnet_name: str, description: str, - subnets: List[Any], + subnets: "List[Subnet]", tags: List[Dict[str, str]], region: str, account_id: str, @@ -1900,18 +1902,16 @@ class RDSBackend(BaseBackend): self.subnet_groups[subnet_name] = subnet_group return subnet_group - def describe_db_subnet_groups( - self, subnet_group_name: str - ) -> Iterable[SubnetGroup]: + def describe_db_subnet_groups(self, subnet_group_name: str) -> List[SubnetGroup]: if subnet_group_name: if subnet_group_name in self.subnet_groups: return [self.subnet_groups[subnet_group_name]] else: raise DBSubnetGroupNotFoundError(subnet_group_name) - return self.subnet_groups.values() + return list(self.subnet_groups.values()) def modify_db_subnet_group( - self, subnet_name: str, description: str, subnets: List[str] + self, subnet_name: str, description: str, subnets: "List[Subnet]" ) -> SubnetGroup: subnet_group = self.subnet_groups.pop(subnet_name) if not subnet_group: @@ -2895,7 +2895,7 @@ class DBParameterGroup(CloudFormationModel): def delete(self, account_id: str, region_name: str) -> None: backend = rds_backends[account_id][region_name] - backend.delete_db_parameter_group(self.name) + backend.delete_db_parameter_group(self.name) # type: ignore[arg-type] @staticmethod def cloudformation_name_type() -> str: diff --git a/moto/route53/models.py b/moto/route53/models.py index 5b943550f..a2fe90393 100644 --- a/moto/route53/models.py +++ b/moto/route53/models.py @@ -291,7 +291,7 @@ class RecordSet(CloudFormationModel): else None ) if not hosted_zone: - hosted_zone = backend.get_hosted_zone(self.hosted_zone_id) + hosted_zone = backend.get_hosted_zone(self.hosted_zone_id) # type: ignore[arg-type] hosted_zone.delete_rrset({"Name": self.name, "Type": self.type_}) @@ -470,8 +470,7 @@ class RecordSetGroup(CloudFormationModel): for record_set in record_sets: hosted_zone.add_rrset(record_set) - record_set_group = RecordSetGroup(hosted_zone.id, record_sets) - return record_set_group + return RecordSetGroup(hosted_zone.id, record_sets) class QueryLoggingConfig(BaseModel): diff --git a/moto/route53resolver/models.py b/moto/route53resolver/models.py index 60a98db7c..bb78713a2 100644 --- a/moto/route53resolver/models.py +++ b/moto/route53resolver/models.py @@ -311,7 +311,7 @@ class ResolverEndpoint(BaseModel): # pylint: disable=too-many-instance-attribut description=f"Route 53 Resolver: {self.id}:{eni_id}", group_ids=self.security_group_ids, interface_type="interface", - private_ip_address=value.get("Ip"), + private_ip_address=value.get("Ip"), # type: ignore[arg-type] private_ip_addresses=[ {"Primary": True, "PrivateIpAddress": value.get("Ip")} ], @@ -515,7 +515,7 @@ class Route53ResolverBackend(BaseBackend): subnet_info = self.ec2_backend.describe_subnets( subnet_ids=[x["SubnetId"]] )[0] - x["Ip"] = subnet_info.get_available_subnet_ip(self) + x["Ip"] = subnet_info.get_available_subnet_ip(self) # type: ignore[arg-type] self._verify_subnet_ips(ip_addresses) self._verify_security_group_ids(security_group_ids) @@ -908,9 +908,9 @@ class Route53ResolverBackend(BaseBackend): if not value.get("Ip"): subnet_info = self.ec2_backend.describe_subnets( - subnet_ids=[value.get("SubnetId")] + subnet_ids=[value.get("SubnetId")] # type: ignore[list-item] )[0] - value["Ip"] = subnet_info.get_available_subnet_ip(self) + value["Ip"] = subnet_info.get_available_subnet_ip(self) # type: ignore[arg-type] self._verify_subnet_ips([value], False) resolver_endpoint.associate_ip_address(value) diff --git a/moto/s3/models.py b/moto/s3/models.py index 07838192d..cde316dfd 100644 --- a/moto/s3/models.py +++ b/moto/s3/models.py @@ -2633,7 +2633,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider): ] -class S3BackendDict(BackendDict): +class S3BackendDict(BackendDict[S3Backend]): """ Encapsulation class to hold S3 backends. diff --git a/moto/sagemaker/utils.py b/moto/sagemaker/utils.py index de5cb873e..ae651eb1c 100644 --- a/moto/sagemaker/utils.py +++ b/moto/sagemaker/utils.py @@ -46,7 +46,7 @@ def load_pipeline_definition_from_s3( bucket_name=pipeline_definition_s3_location["Bucket"], key_name=pipeline_definition_s3_location["ObjectKey"], ) - return json.loads(result.value) + return json.loads(result.value) # type: ignore[union-attr] def arn_formatter(_type: str, _id: str, account_id: str, region_name: str) -> str: diff --git a/moto/ses/models.py b/moto/ses/models.py index 06d0be9f8..f99452551 100644 --- a/moto/ses/models.py +++ b/moto/ses/models.py @@ -332,7 +332,8 @@ class SESBackend(BaseBackend): message = self.__generate_feedback__(msg_type) if message: sns_backends[self.account_id][self.region_name].publish( - message, arn=sns_topic + message, # type: ignore[arg-type] + arn=sns_topic, ) def send_raw_email( diff --git a/moto/sqs/models.py b/moto/sqs/models.py index 37e241bf3..9c01f5f79 100644 --- a/moto/sqs/models.py +++ b/moto/sqs/models.py @@ -467,10 +467,10 @@ class Queue(CloudFormationModel): sqs_backend = sqs_backends[account_id][region_name] queue = sqs_backend.get_queue(queue_name) if "VisibilityTimeout" in properties: - queue.visibility_timeout = int(properties["VisibilityTimeout"]) + queue.visibility_timeout = int(properties["VisibilityTimeout"]) # type: ignore[attr-defined] if "ReceiveMessageWaitTimeSeconds" in properties: - queue.receive_message_wait_time_seconds = int( + queue.receive_message_wait_time_seconds = int( # type: ignore[attr-defined] properties["ReceiveMessageWaitTimeSeconds"] ) return queue @@ -593,15 +593,16 @@ class Queue(CloudFormationModel): ) if result: - [backend.delete_message(self.name, m.receipt_handle) for m in messages] + for m in messages: + backend.delete_message(self.name, m.receipt_handle) # type: ignore[arg-type] else: # Make messages visible again - [ + for m in messages: backend.change_message_visibility( - self.name, m.receipt_handle, visibility_timeout=0 + self.name, + m.receipt_handle, # type: ignore[arg-type] + visibility_timeout=0, ) - for m in messages - ] def delete_message(self, receipt_handle: str) -> None: if receipt_handle in self.deleted_messages: diff --git a/moto/ssm/models.py b/moto/ssm/models.py index dcb0b1e50..de2c638b4 100644 --- a/moto/ssm/models.py +++ b/moto/ssm/models.py @@ -141,7 +141,7 @@ class ParameterDict(DefaultDict[str, List["Parameter"]]): secrets_backend.get_secret_value( secret_name, version_id=version_id, - version_stage=None, + version_stage=None, # type: ignore[arg-type] ) for version_id in sorted_version_ids ] @@ -149,7 +149,7 @@ class ParameterDict(DefaultDict[str, List["Parameter"]]): Parameter( account_id=self.account_id, name=secret["Name"], - value=val.get("SecretString"), + value=val.get("SecretString"), # type: ignore[arg-type] parameter_type="SecureString", description=secret.get("Description"), allowed_pattern=None, diff --git a/setup.cfg b/setup.cfg index ec63fa0d6..6408e97db 100644 --- a/setup.cfg +++ b/setup.cfg @@ -265,7 +265,7 @@ markers = relative_files = True [flake8] -ignore = W503,W605,E128,E501,E203,E266,E501,E231,FS003 +ignore = W503,W605,E128,E501,E203,E266,E501,E231,FS003,E704 exclude = moto/packages,dist,tests/terraformtests [pylint.MASTER]