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

Techdebt: typing for backends (#7069)

Browse Source
This commit is contained in:
tungol 2023-12-01 03:51:28 -08:00 committed by GitHub
parent 4e1ea0bb6f
commit 8a16a6a862
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
32 changed files with 617 additions and 87 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
moto/awslambda/models.py
View File

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

2
moto/awslambda_simple/responses.py
View File

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

535
moto/backends.py
View File

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

10
moto/batch/models.py
View File

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

2
moto/batch_simple/responses.py
View File

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

9
moto/cloudformation/parsing.py
View File

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

2
moto/cloudformation/utils.py
View File

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

2
moto/cognitoidp/responses.py
View File

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

2
moto/core/__init__.py
View File

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

21
moto/core/base_backend.py
View File

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

8
moto/core/models.py
View File

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

4
moto/ds/models.py
View File

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

2
moto/ec2/models/__init__.py
View File

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

3
moto/ec2/models/elastic_block_store.py
View File

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

2
moto/ec2/models/instances.py
View File

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

5
moto/ec2/models/subnets.py
View File

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

2
moto/elbv2/models.py
View File

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

4
moto/events/models.py
View File

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

4
moto/events/notifications.py
View File

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

2
moto/iam/models.py
View File

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

2
moto/iot/models.py
View File

@ -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", [])]
)

2
moto/logs/models.py
View File

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

8
moto/moto_server/werkzeug_app.py
View File

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

26
moto/rds/models.py
View File

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

5
moto/route53/models.py
View File

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

8
moto/route53resolver/models.py
View File

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

2
moto/s3/models.py
View File

@ -2633,7 +2633,7 @@ class S3Backend(BaseBackend, CloudWatchMetricProvider):
]
class S3BackendDict(BackendDict):
class S3BackendDict(BackendDict[S3Backend]):
"""
Encapsulation class to hold S3 backends.

2
moto/sagemaker/utils.py
View File

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

3
moto/ses/models.py
View File

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

15
moto/sqs/models.py
View File

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

4
moto/ssm/models.py
View File

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

2
setup.cfg
View File

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