Skip to content

ireally drop python<=3.7 support #3678

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
wants to merge 1 commit into
base: develop
Choose a base branch
from
Open
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
@@ -66,7 +66,7 @@ def test_function_with_all_event_types(self):
# assert LambdaEventSourceMappings
event_source_mappings = lambda_client.list_event_source_mappings()["EventSourceMappings"]
event_source_mapping_configurations = [x for x in event_source_mappings if x["FunctionArn"] == alias_arn]
event_source_mapping_arns = set([x["EventSourceArn"] for x in event_source_mapping_configurations])
event_source_mapping_arns = {x["EventSourceArn"] for x in event_source_mapping_configurations}

kinesis_client = self.client_provider.kinesis_client
kinesis_stream_name = self.get_physical_id_by_type("AWS::Kinesis::Stream")
2 changes: 1 addition & 1 deletion integration/helpers/request_utils.py
Original file line number Diff line number Diff line change
@@ -34,4 +34,4 @@ def _normalize_response_headers(self):
# Need to check for response is None here since the __bool__ method checks 200 <= status < 400
return {}

return dict((k.lower(), v) for k, v in self.response.headers.items())
return {k.lower(): v for k, v in self.response.headers.items()}
1 change: 0 additions & 1 deletion integration/single/__init__.py
Original file line number Diff line number Diff line change
@@ -1 +0,0 @@

250 changes: 125 additions & 125 deletions samtranslator/internal/schema_source/aws_serverless_api.py

Large diffs are not rendered by default.

Original file line number Diff line number Diff line change
@@ -23,19 +23,19 @@ class Location(BaseModel):


class Properties(BaseModel):
Location: Union[str, Location] = properties("Location")
NotificationARNs: Optional[PassThroughProp] = passthrough_prop(
Location: str | Location = properties("Location")
NotificationARNs: PassThroughProp | None = passthrough_prop(
PROPERTIES_STEM,
"NotificationARNs",
["AWS::CloudFormation::Stack", "Properties", "NotificationARNs"],
)
Parameters: Optional[PassThroughProp] = passthrough_prop(
Parameters: PassThroughProp | None = passthrough_prop(
PROPERTIES_STEM,
"Parameters",
["AWS::CloudFormation::Stack", "Properties", "Parameters"],
)
Tags: Optional[Dict[str, Any]] = properties("Tags")
TimeoutInMinutes: Optional[PassThroughProp] = passthrough_prop(
Tags: dict[str, Any] | None = properties("Tags")
TimeoutInMinutes: PassThroughProp | None = passthrough_prop(
PROPERTIES_STEM,
"TimeoutInMinutes",
["AWS::CloudFormation::Stack", "Properties", "TimeoutInMinutes"],
548 changes: 274 additions & 274 deletions samtranslator/internal/schema_source/aws_serverless_function.py

Large diffs are not rendered by default.

140 changes: 70 additions & 70 deletions samtranslator/internal/schema_source/aws_serverless_httpapi.py
Original file line number Diff line number Diff line change
@@ -24,80 +24,80 @@


class OAuth2Authorizer(BaseModel):
AuthorizationScopes: Optional[List[str]] = oauth2authorizer("AuthorizationScopes")
IdentitySource: Optional[str] = oauth2authorizer("IdentitySource")
JwtConfiguration: Optional[PassThroughProp] = oauth2authorizer("JwtConfiguration")
AuthorizationScopes: list[str] | None = oauth2authorizer("AuthorizationScopes")
IdentitySource: str | None = oauth2authorizer("IdentitySource")
JwtConfiguration: PassThroughProp | None = oauth2authorizer("JwtConfiguration")


class LambdaAuthorizerIdentity(BaseModel):
Context: Optional[List[str]] = lambdauthorizeridentity("Context")
Headers: Optional[List[str]] = lambdauthorizeridentity("Headers")
QueryStrings: Optional[List[str]] = lambdauthorizeridentity("QueryStrings")
ReauthorizeEvery: Optional[int] = lambdauthorizeridentity("ReauthorizeEvery")
StageVariables: Optional[List[str]] = lambdauthorizeridentity("StageVariables")
Context: list[str] | None = lambdauthorizeridentity("Context")
Headers: list[str] | None = lambdauthorizeridentity("Headers")
QueryStrings: list[str] | None = lambdauthorizeridentity("QueryStrings")
ReauthorizeEvery: int | None = lambdauthorizeridentity("ReauthorizeEvery")
StageVariables: list[str] | None = lambdauthorizeridentity("StageVariables")


class LambdaAuthorizer(BaseModel):
# TODO: Many tests use floats for the version string; docs only mention string
AuthorizerPayloadFormatVersion: Union[Literal["1.0", "2.0"], float] = lambdaauthorizer(
AuthorizerPayloadFormatVersion: Literal["1.0", "2.0"] | float = lambdaauthorizer(
"AuthorizerPayloadFormatVersion"
)
EnableSimpleResponses: Optional[bool] = lambdaauthorizer("EnableSimpleResponses")
EnableSimpleResponses: bool | None = lambdaauthorizer("EnableSimpleResponses")
FunctionArn: SamIntrinsicable[str] = lambdaauthorizer("FunctionArn")
FunctionInvokeRole: Optional[SamIntrinsicable[str]] = lambdaauthorizer("FunctionInvokeRole")
EnableFunctionDefaultPermissions: Optional[bool] # TODO: add docs
Identity: Optional[LambdaAuthorizerIdentity] = lambdaauthorizer("Identity")
FunctionInvokeRole: SamIntrinsicable[str] | None = lambdaauthorizer("FunctionInvokeRole")
EnableFunctionDefaultPermissions: bool | None # TODO: add docs
Identity: LambdaAuthorizerIdentity | None = lambdaauthorizer("Identity")


class Auth(BaseModel):
# TODO: Docs doesn't say it's a map
Authorizers: Optional[
Dict[
Authorizers: None | (
dict[
str,
Union[
OAuth2Authorizer,
LambdaAuthorizer,
],
(
OAuth2Authorizer |
LambdaAuthorizer
),
]
] = auth("Authorizers")
DefaultAuthorizer: Optional[str] = auth("DefaultAuthorizer")
EnableIamAuthorizer: Optional[bool] = auth("EnableIamAuthorizer")
) = auth("Authorizers")
DefaultAuthorizer: str | None = auth("DefaultAuthorizer")
EnableIamAuthorizer: bool | None = auth("EnableIamAuthorizer")


class CorsConfiguration(BaseModel):
AllowCredentials: Optional[bool] = corsconfiguration("AllowCredentials")
AllowHeaders: Optional[List[str]] = corsconfiguration("AllowHeaders")
AllowMethods: Optional[List[str]] = corsconfiguration("AllowMethods")
AllowOrigins: Optional[List[str]] = corsconfiguration("AllowOrigins")
ExposeHeaders: Optional[List[str]] = corsconfiguration("ExposeHeaders")
MaxAge: Optional[int] = corsconfiguration("MaxAge")
AllowCredentials: bool | None = corsconfiguration("AllowCredentials")
AllowHeaders: list[str] | None = corsconfiguration("AllowHeaders")
AllowMethods: list[str] | None = corsconfiguration("AllowMethods")
AllowOrigins: list[str] | None = corsconfiguration("AllowOrigins")
ExposeHeaders: list[str] | None = corsconfiguration("ExposeHeaders")
MaxAge: int | None = corsconfiguration("MaxAge")


class DefinitionUri(BaseModel):
Bucket: str = definitionuri("Bucket")
Key: str = definitionuri("Key")
Version: Optional[str] = definitionuri("Version")
Version: str | None = definitionuri("Version")


class Route53(BaseModel):
DistributionDomainName: Optional[PassThroughProp] = route53("DistributionDomainName")
EvaluateTargetHealth: Optional[PassThroughProp] = route53("EvaluateTargetHealth")
HostedZoneId: Optional[PassThroughProp] = route53("HostedZoneId")
HostedZoneName: Optional[PassThroughProp] = route53("HostedZoneName")
IpV6: Optional[bool] = route53("IpV6")
SetIdentifier: Optional[PassThroughProp] # TODO: add docs
Region: Optional[PassThroughProp] # TODO: add docs
DistributionDomainName: PassThroughProp | None = route53("DistributionDomainName")
EvaluateTargetHealth: PassThroughProp | None = route53("EvaluateTargetHealth")
HostedZoneId: PassThroughProp | None = route53("HostedZoneId")
HostedZoneName: PassThroughProp | None = route53("HostedZoneName")
IpV6: bool | None = route53("IpV6")
SetIdentifier: PassThroughProp | None # TODO: add docs
Region: PassThroughProp | None # TODO: add docs


class Domain(BaseModel):
BasePath: Optional[List[str]] = domain("BasePath")
BasePath: list[str] | None = domain("BasePath")
CertificateArn: PassThroughProp = domain("CertificateArn")
DomainName: PassThroughProp = domain("DomainName")
EndpointConfiguration: Optional[SamIntrinsicable[Literal["REGIONAL"]]] = domain("EndpointConfiguration")
MutualTlsAuthentication: Optional[PassThroughProp] = domain("MutualTlsAuthentication")
OwnershipVerificationCertificateArn: Optional[PassThroughProp] = domain("OwnershipVerificationCertificateArn")
Route53: Optional[Route53] = domain("Route53")
SecurityPolicy: Optional[PassThroughProp] = domain("SecurityPolicy")
EndpointConfiguration: SamIntrinsicable[Literal["REGIONAL"]] | None = domain("EndpointConfiguration")
MutualTlsAuthentication: PassThroughProp | None = domain("MutualTlsAuthentication")
OwnershipVerificationCertificateArn: PassThroughProp | None = domain("OwnershipVerificationCertificateArn")
Route53: Route53 | None = domain("Route53")
SecurityPolicy: PassThroughProp | None = domain("SecurityPolicy")


AccessLogSettings = Optional[PassThroughProp]
@@ -110,39 +110,39 @@ class Domain(BaseModel):


class Properties(BaseModel):
AccessLogSettings: Optional[AccessLogSettings] = properties("AccessLogSettings")
Auth: Optional[Auth] = properties("Auth")
AccessLogSettings: AccessLogSettings | None = properties("AccessLogSettings")
Auth: Auth | None = properties("Auth")
# TODO: Also string like in the docs?
CorsConfiguration: Optional[CorsConfigurationType] = properties("CorsConfiguration")
DefaultRouteSettings: Optional[DefaultRouteSettings] = properties("DefaultRouteSettings")
DefinitionBody: Optional[DictStrAny] = properties("DefinitionBody")
DefinitionUri: Optional[Union[str, DefinitionUri]] = properties("DefinitionUri")
Description: Optional[str] = properties("Description")
DisableExecuteApiEndpoint: Optional[PassThroughProp] = properties("DisableExecuteApiEndpoint")
Domain: Optional[Domain] = properties("Domain")
FailOnWarnings: Optional[FailOnWarnings] = properties("FailOnWarnings")
RouteSettings: Optional[RouteSettings] = properties("RouteSettings")
StageName: Optional[PassThroughProp] = properties("StageName")
StageVariables: Optional[StageVariables] = properties("StageVariables")
Tags: Optional[Tags] = properties("Tags")
PropagateTags: Optional[bool] # TODO: add docs
Name: Optional[PassThroughProp] = properties("Name")
CorsConfiguration: CorsConfigurationType | None = properties("CorsConfiguration")
DefaultRouteSettings: DefaultRouteSettings | None = properties("DefaultRouteSettings")
DefinitionBody: DictStrAny | None = properties("DefinitionBody")
DefinitionUri: str | DefinitionUri | None = properties("DefinitionUri")
Description: str | None = properties("Description")
DisableExecuteApiEndpoint: PassThroughProp | None = properties("DisableExecuteApiEndpoint")
Domain: Domain | None = properties("Domain")
FailOnWarnings: FailOnWarnings | None = properties("FailOnWarnings")
RouteSettings: RouteSettings | None = properties("RouteSettings")
StageName: PassThroughProp | None = properties("StageName")
StageVariables: StageVariables | None = properties("StageVariables")
Tags: Tags | None = properties("Tags")
PropagateTags: bool | None # TODO: add docs
Name: PassThroughProp | None = properties("Name")


class Globals(BaseModel):
Auth: Optional[Auth] = properties("Auth")
AccessLogSettings: Optional[AccessLogSettings] = properties("AccessLogSettings")
StageVariables: Optional[StageVariables] = properties("StageVariables")
Tags: Optional[Tags] = properties("Tags")
RouteSettings: Optional[RouteSettings] = properties("RouteSettings")
FailOnWarnings: Optional[FailOnWarnings] = properties("FailOnWarnings")
Domain: Optional[Domain] = properties("Domain")
CorsConfiguration: Optional[CorsConfigurationType] = properties("CorsConfiguration")
DefaultRouteSettings: Optional[DefaultRouteSettings] = properties("DefaultRouteSettings")
PropagateTags: Optional[bool] # TODO: add docs
Auth: Auth | None = properties("Auth")
AccessLogSettings: AccessLogSettings | None = properties("AccessLogSettings")
StageVariables: StageVariables | None = properties("StageVariables")
Tags: Tags | None = properties("Tags")
RouteSettings: RouteSettings | None = properties("RouteSettings")
FailOnWarnings: FailOnWarnings | None = properties("FailOnWarnings")
Domain: Domain | None = properties("Domain")
CorsConfiguration: CorsConfigurationType | None = properties("CorsConfiguration")
DefaultRouteSettings: DefaultRouteSettings | None = properties("DefaultRouteSettings")
PropagateTags: bool | None # TODO: add docs


class Resource(ResourceAttributes):
Type: Literal["AWS::Serverless::HttpApi"]
Properties: Optional[Properties]
Connectors: Optional[Dict[str, EmbeddedConnector]]
Properties: Properties | None
Connectors: dict[str, EmbeddedConnector] | None
Original file line number Diff line number Diff line change
@@ -29,38 +29,38 @@ class ContentUri(BaseModel):
"Key",
["AWS::Lambda::LayerVersion.Content", "S3Key"],
)
Version: Optional[PassThroughProp] = passthrough_prop(
Version: PassThroughProp | None = passthrough_prop(
CONTENT_URI_STEM,
"Version",
["AWS::Lambda::LayerVersion.Content", "S3ObjectVersion"],
)


class Properties(BaseModel):
CompatibleArchitectures: Optional[PassThroughProp] = passthrough_prop(
CompatibleArchitectures: PassThroughProp | None = passthrough_prop(
PROPERTIES_STEM,
"CompatibleArchitectures",
["AWS::Lambda::LayerVersion", "Properties", "CompatibleArchitectures"],
)
CompatibleRuntimes: Optional[PassThroughProp] = passthrough_prop(
CompatibleRuntimes: PassThroughProp | None = passthrough_prop(
PROPERTIES_STEM,
"CompatibleRuntimes",
["AWS::Lambda::LayerVersion", "Properties", "CompatibleRuntimes"],
)
PublishLambdaVersion: Optional[bool] # TODO: add docs
ContentUri: Union[str, ContentUri] = properties("ContentUri")
Description: Optional[PassThroughProp] = passthrough_prop(
PublishLambdaVersion: bool | None # TODO: add docs
ContentUri: str | ContentUri = properties("ContentUri")
Description: PassThroughProp | None = passthrough_prop(
PROPERTIES_STEM,
"Description",
["AWS::Lambda::LayerVersion", "Properties", "Description"],
)
LayerName: Optional[PassThroughProp] = properties("LayerName")
LicenseInfo: Optional[PassThroughProp] = passthrough_prop(
LayerName: PassThroughProp | None = properties("LayerName")
LicenseInfo: PassThroughProp | None = passthrough_prop(
PROPERTIES_STEM,
"LicenseInfo",
["AWS::Lambda::LayerVersion", "Properties", "LicenseInfo"],
)
RetentionPolicy: Optional[SamIntrinsicable[str]] = properties("RetentionPolicy")
RetentionPolicy: SamIntrinsicable[str] | None = properties("RetentionPolicy")


class Resource(ResourceAttributes):
@@ -69,4 +69,4 @@ class Resource(ResourceAttributes):


class Globals(BaseModel):
PublishLambdaVersion: Optional[bool] # TODO: add docs
PublishLambdaVersion: bool | None # TODO: add docs
Original file line number Diff line number Diff line change
@@ -35,28 +35,28 @@ class PrimaryKey(BaseModel):


class Properties(BaseModel):
PointInTimeRecoverySpecification: Optional[PassThroughProp] # TODO: add docs
PrimaryKey: Optional[PrimaryKey] = properties("PrimaryKey")
ProvisionedThroughput: Optional[PassThroughProp] = passthrough_prop(
PointInTimeRecoverySpecification: PassThroughProp | None # TODO: add docs
PrimaryKey: PrimaryKey | None = properties("PrimaryKey")
ProvisionedThroughput: PassThroughProp | None = passthrough_prop(
PROPERTIES_STEM,
"ProvisionedThroughput",
["AWS::DynamoDB::Table", "Properties", "ProvisionedThroughput"],
)
SSESpecification: Optional[SSESpecification] = passthrough_prop(
SSESpecification: SSESpecification | None = passthrough_prop(
PROPERTIES_STEM,
"SSESpecification",
["AWS::DynamoDB::Table", "Properties", "SSESpecification"],
)
TableName: Optional[PassThroughProp] = passthrough_prop(
TableName: PassThroughProp | None = passthrough_prop(
PROPERTIES_STEM,
"TableName",
["AWS::DynamoDB::Table", "Properties", "TableName"],
)
Tags: Optional[Dict[str, Any]] = properties("Tags")
Tags: dict[str, Any] | None = properties("Tags")


class Globals(BaseModel):
SSESpecification: Optional[SSESpecification] = passthrough_prop(
SSESpecification: SSESpecification | None = passthrough_prop(
PROPERTIES_STEM,
"SSESpecification",
["AWS::DynamoDB::Table", "Properties", "SSESpecification"],
@@ -65,5 +65,5 @@ class Globals(BaseModel):

class Resource(ResourceAttributes):
Type: Literal["AWS::Serverless::SimpleTable"]
Properties: Optional[Properties]
Connectors: Optional[Dict[str, EmbeddedConnector]]
Properties: Properties | None
Connectors: dict[str, EmbeddedConnector] | None
176 changes: 88 additions & 88 deletions samtranslator/internal/schema_source/aws_serverless_statemachine.py
Original file line number Diff line number Diff line change
@@ -28,26 +28,26 @@


class DeadLetterConfig(BaseModel):
Arn: Optional[PassThroughProp] = deadletterconfig("Arn")
QueueLogicalId: Optional[str] = deadletterconfig("QueueLogicalId")
Type: Optional[Literal["SQS"]] = deadletterconfig("Type")
Arn: PassThroughProp | None = deadletterconfig("Arn")
QueueLogicalId: str | None = deadletterconfig("QueueLogicalId")
Type: Literal["SQS"] | None = deadletterconfig("Type")


class ScheduleTarget(BaseModel):
Id: PassThroughProp = scheduletarget("Id")


class ScheduleEventProperties(BaseModel):
DeadLetterConfig: Optional[DeadLetterConfig] = scheduleeventproperties("DeadLetterConfig")
Description: Optional[PassThroughProp] = scheduleeventproperties("Description")
Enabled: Optional[bool] = scheduleeventproperties("Enabled")
Input: Optional[PassThroughProp] = scheduleeventproperties("Input")
Name: Optional[PassThroughProp] = scheduleeventproperties("Name")
RetryPolicy: Optional[PassThroughProp] = scheduleeventproperties("RetryPolicy")
Schedule: Optional[PassThroughProp] = scheduleeventproperties("Schedule")
State: Optional[PassThroughProp] = scheduleeventproperties("State")
Target: Optional[ScheduleTarget] = scheduleeventproperties("Target")
RoleArn: Optional[PassThroughProp] # TODO: add doc
DeadLetterConfig: DeadLetterConfig | None = scheduleeventproperties("DeadLetterConfig")
Description: PassThroughProp | None = scheduleeventproperties("Description")
Enabled: bool | None = scheduleeventproperties("Enabled")
Input: PassThroughProp | None = scheduleeventproperties("Input")
Name: PassThroughProp | None = scheduleeventproperties("Name")
RetryPolicy: PassThroughProp | None = scheduleeventproperties("RetryPolicy")
Schedule: PassThroughProp | None = scheduleeventproperties("Schedule")
State: PassThroughProp | None = scheduleeventproperties("State")
Target: ScheduleTarget | None = scheduleeventproperties("Target")
RoleArn: PassThroughProp | None # TODO: add doc


class ScheduleEvent(BaseModel):
@@ -56,22 +56,22 @@ class ScheduleEvent(BaseModel):


class ScheduleV2EventProperties(BaseModel):
DeadLetterConfig: Optional[DeadLetterConfig] = scheduleeventv2properties("DeadLetterConfig")
Description: Optional[PassThroughProp] = scheduleeventv2properties("Description")
EndDate: Optional[PassThroughProp] = scheduleeventv2properties("EndDate")
FlexibleTimeWindow: Optional[PassThroughProp] = scheduleeventv2properties("FlexibleTimeWindow")
GroupName: Optional[PassThroughProp] = scheduleeventv2properties("GroupName")
Input: Optional[PassThroughProp] = scheduleeventv2properties("Input")
KmsKeyArn: Optional[PassThroughProp] = scheduleeventv2properties("KmsKeyArn")
Name: Optional[PassThroughProp] = scheduleeventv2properties("Name")
PermissionsBoundary: Optional[PassThroughProp] = scheduleeventv2properties("PermissionsBoundary")
RetryPolicy: Optional[PassThroughProp] = scheduleeventv2properties("RetryPolicy")
RoleArn: Optional[PassThroughProp] = scheduleeventv2properties("RoleArn")
ScheduleExpression: Optional[PassThroughProp] = scheduleeventv2properties("ScheduleExpression")
ScheduleExpressionTimezone: Optional[PassThroughProp] = scheduleeventv2properties("ScheduleExpressionTimezone")
StartDate: Optional[PassThroughProp] = scheduleeventv2properties("StartDate")
State: Optional[PassThroughProp] = scheduleeventv2properties("State")
OmitName: Optional[bool] # TODO: add doc
DeadLetterConfig: DeadLetterConfig | None = scheduleeventv2properties("DeadLetterConfig")
Description: PassThroughProp | None = scheduleeventv2properties("Description")
EndDate: PassThroughProp | None = scheduleeventv2properties("EndDate")
FlexibleTimeWindow: PassThroughProp | None = scheduleeventv2properties("FlexibleTimeWindow")
GroupName: PassThroughProp | None = scheduleeventv2properties("GroupName")
Input: PassThroughProp | None = scheduleeventv2properties("Input")
KmsKeyArn: PassThroughProp | None = scheduleeventv2properties("KmsKeyArn")
Name: PassThroughProp | None = scheduleeventv2properties("Name")
PermissionsBoundary: PassThroughProp | None = scheduleeventv2properties("PermissionsBoundary")
RetryPolicy: PassThroughProp | None = scheduleeventv2properties("RetryPolicy")
RoleArn: PassThroughProp | None = scheduleeventv2properties("RoleArn")
ScheduleExpression: PassThroughProp | None = scheduleeventv2properties("ScheduleExpression")
ScheduleExpressionTimezone: PassThroughProp | None = scheduleeventv2properties("ScheduleExpressionTimezone")
StartDate: PassThroughProp | None = scheduleeventv2properties("StartDate")
State: PassThroughProp | None = scheduleeventv2properties("State")
OmitName: bool | None # TODO: add doc


class ScheduleV2Event(BaseModel):
@@ -80,24 +80,24 @@ class ScheduleV2Event(BaseModel):


class ResourcePolicy(BaseModel):
AwsAccountBlacklist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("AwsAccountBlacklist")
AwsAccountWhitelist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("AwsAccountWhitelist")
CustomStatements: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("CustomStatements")
IntrinsicVpcBlacklist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("IntrinsicVpcBlacklist")
IntrinsicVpcWhitelist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("IntrinsicVpcWhitelist")
IntrinsicVpceBlacklist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("IntrinsicVpceBlacklist")
IntrinsicVpceWhitelist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("IntrinsicVpceWhitelist")
IpRangeBlacklist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("IpRangeBlacklist")
IpRangeWhitelist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("IpRangeWhitelist")
SourceVpcBlacklist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("SourceVpcBlacklist")
SourceVpcWhitelist: Optional[List[Union[str, DictStrAny]]] = resourcepolicy("SourceVpcWhitelist")
AwsAccountBlacklist: list[str | DictStrAny] | None = resourcepolicy("AwsAccountBlacklist")
AwsAccountWhitelist: list[str | DictStrAny] | None = resourcepolicy("AwsAccountWhitelist")
CustomStatements: list[str | DictStrAny] | None = resourcepolicy("CustomStatements")
IntrinsicVpcBlacklist: list[str | DictStrAny] | None = resourcepolicy("IntrinsicVpcBlacklist")
IntrinsicVpcWhitelist: list[str | DictStrAny] | None = resourcepolicy("IntrinsicVpcWhitelist")
IntrinsicVpceBlacklist: list[str | DictStrAny] | None = resourcepolicy("IntrinsicVpceBlacklist")
IntrinsicVpceWhitelist: list[str | DictStrAny] | None = resourcepolicy("IntrinsicVpceWhitelist")
IpRangeBlacklist: list[str | DictStrAny] | None = resourcepolicy("IpRangeBlacklist")
IpRangeWhitelist: list[str | DictStrAny] | None = resourcepolicy("IpRangeWhitelist")
SourceVpcBlacklist: list[str | DictStrAny] | None = resourcepolicy("SourceVpcBlacklist")
SourceVpcWhitelist: list[str | DictStrAny] | None = resourcepolicy("SourceVpcWhitelist")


class CloudWatchEventProperties(BaseModel):
EventBusName: Optional[PassThroughProp] = cloudwatcheventproperties("EventBusName")
Input: Optional[PassThroughProp] = cloudwatcheventproperties("Input")
InputPath: Optional[PassThroughProp] = cloudwatcheventproperties("InputPath")
Pattern: Optional[PassThroughProp] = cloudwatcheventproperties("Pattern")
EventBusName: PassThroughProp | None = cloudwatcheventproperties("EventBusName")
Input: PassThroughProp | None = cloudwatcheventproperties("Input")
InputPath: PassThroughProp | None = cloudwatcheventproperties("InputPath")
Pattern: PassThroughProp | None = cloudwatcheventproperties("Pattern")


class CloudWatchEvent(BaseModel):
@@ -110,15 +110,15 @@ class EventBridgeRuleTarget(BaseModel):


class EventBridgeRuleEventProperties(BaseModel):
DeadLetterConfig: Optional[DeadLetterConfig] = eventbridgeruleeventproperties("DeadLetterConfig")
EventBusName: Optional[PassThroughProp] = eventbridgeruleeventproperties("EventBusName")
Input: Optional[PassThroughProp] = eventbridgeruleeventproperties("Input")
InputPath: Optional[PassThroughProp] = eventbridgeruleeventproperties("InputPath")
Pattern: Optional[PassThroughProp] = eventbridgeruleeventproperties("Pattern")
RetryPolicy: Optional[PassThroughProp] = eventbridgeruleeventproperties("RetryPolicy")
Target: Optional[EventBridgeRuleTarget] = eventbridgeruleeventproperties("Target")
RuleName: Optional[PassThroughProp] = eventbridgeruleeventproperties("RuleName")
InputTransformer: Optional[PassThroughProp] # TODO: add docs
DeadLetterConfig: DeadLetterConfig | None = eventbridgeruleeventproperties("DeadLetterConfig")
EventBusName: PassThroughProp | None = eventbridgeruleeventproperties("EventBusName")
Input: PassThroughProp | None = eventbridgeruleeventproperties("Input")
InputPath: PassThroughProp | None = eventbridgeruleeventproperties("InputPath")
Pattern: PassThroughProp | None = eventbridgeruleeventproperties("Pattern")
RetryPolicy: PassThroughProp | None = eventbridgeruleeventproperties("RetryPolicy")
Target: EventBridgeRuleTarget | None = eventbridgeruleeventproperties("Target")
RuleName: PassThroughProp | None = eventbridgeruleeventproperties("RuleName")
InputTransformer: PassThroughProp | None # TODO: add docs


class EventBridgeRuleEvent(BaseModel):
@@ -127,18 +127,18 @@ class EventBridgeRuleEvent(BaseModel):


class Auth(BaseModel):
ApiKeyRequired: Optional[bool] = apiauth("ApiKeyRequired")
AuthorizationScopes: Optional[List[str]] = apiauth("AuthorizationScopes")
Authorizer: Optional[str] = apiauth("Authorizer")
ResourcePolicy: Optional[ResourcePolicy] = apiauth("ResourcePolicy")
ApiKeyRequired: bool | None = apiauth("ApiKeyRequired")
AuthorizationScopes: list[str] | None = apiauth("AuthorizationScopes")
Authorizer: str | None = apiauth("Authorizer")
ResourcePolicy: ResourcePolicy | None = apiauth("ResourcePolicy")


class ApiEventProperties(BaseModel):
Auth: Optional[Auth] = apieventproperties("Auth")
Auth: Auth | None = apieventproperties("Auth")
Method: str = apieventproperties("Method")
Path: str = apieventproperties("Path")
RestApiId: Optional[SamIntrinsicable[str]] = apieventproperties("RestApiId")
UnescapeMappingTemplate: Optional[bool] = apieventproperties("UnescapeMappingTemplate")
RestApiId: SamIntrinsicable[str] | None = apieventproperties("RestApiId")
UnescapeMappingTemplate: bool | None = apieventproperties("UnescapeMappingTemplate")


class ApiEvent(BaseModel):
@@ -147,41 +147,41 @@ class ApiEvent(BaseModel):


class Properties(BaseModel):
Definition: Optional[DictStrAny] = properties("Definition")
DefinitionSubstitutions: Optional[DictStrAny] = properties("DefinitionSubstitutions")
DefinitionUri: Optional[Union[str, PassThroughProp]] = properties("DefinitionUri")
Events: Optional[
Dict[
Definition: DictStrAny | None = properties("Definition")
DefinitionSubstitutions: DictStrAny | None = properties("DefinitionSubstitutions")
DefinitionUri: str | PassThroughProp | None = properties("DefinitionUri")
Events: None | (
dict[
str,
Union[
ScheduleEvent,
ScheduleV2Event,
CloudWatchEvent,
EventBridgeRuleEvent,
ApiEvent,
],
(
ScheduleEvent |
ScheduleV2Event |
CloudWatchEvent |
EventBridgeRuleEvent |
ApiEvent
),
]
] = properties("Events")
Logging: Optional[PassThroughProp] = properties("Logging")
Name: Optional[PassThroughProp] = properties("Name")
PermissionsBoundary: Optional[PassThroughProp] = properties("PermissionsBoundary")
Policies: Optional[Union[str, DictStrAny, List[Union[str, DictStrAny]]]] = properties("Policies")
Role: Optional[PassThroughProp] = properties("Role")
RolePath: Optional[PassThroughProp] = properties("RolePath")
Tags: Optional[DictStrAny] = properties("Tags")
PropagateTags: Optional[bool] # TODO: add docs
Tracing: Optional[PassThroughProp] = properties("Tracing")
Type: Optional[PassThroughProp] = properties("Type")
AutoPublishAlias: Optional[PassThroughProp]
DeploymentPreference: Optional[PassThroughProp]
UseAliasAsEventTarget: Optional[bool]
) = properties("Events")
Logging: PassThroughProp | None = properties("Logging")
Name: PassThroughProp | None = properties("Name")
PermissionsBoundary: PassThroughProp | None = properties("PermissionsBoundary")
Policies: str | DictStrAny | list[str | DictStrAny] | None = properties("Policies")
Role: PassThroughProp | None = properties("Role")
RolePath: PassThroughProp | None = properties("RolePath")
Tags: DictStrAny | None = properties("Tags")
PropagateTags: bool | None # TODO: add docs
Tracing: PassThroughProp | None = properties("Tracing")
Type: PassThroughProp | None = properties("Type")
AutoPublishAlias: PassThroughProp | None
DeploymentPreference: PassThroughProp | None
UseAliasAsEventTarget: bool | None


class Resource(ResourceAttributes):
Type: Literal["AWS::Serverless::StateMachine"]
Properties: Properties
Connectors: Optional[Dict[str, EmbeddedConnector]]
Connectors: dict[str, EmbeddedConnector] | None


class Globals(BaseModel):
PropagateTags: Optional[bool] # TODO: add docs
PropagateTags: bool | None # TODO: add docs
38 changes: 19 additions & 19 deletions samtranslator/internal/schema_source/schema.py
Original file line number Diff line number Diff line change
@@ -23,12 +23,12 @@


class Globals(BaseModel):
Function: Optional[aws_serverless_function.Globals]
Api: Optional[aws_serverless_api.Globals]
HttpApi: Optional[aws_serverless_httpapi.Globals]
SimpleTable: Optional[aws_serverless_simpletable.Globals]
StateMachine: Optional[aws_serverless_statemachine.Globals]
LayerVersion: Optional[aws_serverless_layerversion.Globals]
Function: aws_serverless_function.Globals | None
Api: aws_serverless_api.Globals | None
HttpApi: aws_serverless_httpapi.Globals | None
SimpleTable: aws_serverless_simpletable.Globals | None
StateMachine: aws_serverless_statemachine.Globals | None
LayerVersion: aws_serverless_layerversion.Globals | None


Resources = Union[
@@ -45,25 +45,25 @@ class Globals(BaseModel):


class _ModelWithoutResources(LenientBaseModel):
Globals: Optional[Globals]
Globals: Globals | None


class SamModel(_ModelWithoutResources):
Resources: Dict[
Resources: dict[
str,
Union[
Resources,
(
Resources |
# Ignore resources that are not AWS::Serverless::*
any_cfn_resource.Resource,
],
any_cfn_resource.Resource
),
]


class Model(_ModelWithoutResources):
Resources: Dict[str, Resources]
Resources: dict[str, Resources]


def get_schema(model: Type[pydantic.BaseModel]) -> Dict[str, Any]:
def get_schema(model: type[pydantic.BaseModel]) -> dict[str, Any]:
obj = model.schema()

# http://json-schema.org/understanding-json-schema/reference/schema.html#schema
@@ -83,7 +83,7 @@ def json_dumps(obj: Any) -> str:
return json.dumps(obj, indent=2, sort_keys=True) + "\n"


def _replace_in_dict(d: Dict[str, Any], keyword: str, replace: Callable[[Dict[str, Any]], Any]) -> Dict[str, Any]:
def _replace_in_dict(d: dict[str, Any], keyword: str, replace: Callable[[dict[str, Any]], Any]) -> dict[str, Any]:
"""
Replace any dict containing keyword.
@@ -97,7 +97,7 @@ def _replace_in_dict(d: Dict[str, Any], keyword: str, replace: Callable[[Dict[st
return d


def _deep_get(d: Dict[str, Any], path: List[str]) -> Dict[str, Any]:
def _deep_get(d: dict[str, Any], path: list[str]) -> dict[str, Any]:
"""
Returns value at path defined by the keys in `path`.
"""
@@ -106,7 +106,7 @@ def _deep_get(d: Dict[str, Any], path: List[str]) -> Dict[str, Any]:
return d


def _add_embedded_connectors(schema: Dict[str, Any]) -> None:
def _add_embedded_connectors(schema: dict[str, Any]) -> None:
"""
Add embedded Connectors resource attribute to supported CloudFormation resources.
"""
@@ -123,7 +123,7 @@ def _add_embedded_connectors(schema: Dict[str, Any]) -> None:
schema["definitions"][resource]["properties"]["Connectors"] = embedded_connector


def extend_with_cfn_schema(sam_schema: Dict[str, Any], cfn_schema: Dict[str, Any]) -> None:
def extend_with_cfn_schema(sam_schema: dict[str, Any], cfn_schema: dict[str, Any]) -> None:
"""
Add CloudFormation resources and template syntax to SAM schema.
"""
@@ -152,7 +152,7 @@ def extend_with_cfn_schema(sam_schema: Dict[str, Any], cfn_schema: Dict[str, Any
_add_embedded_connectors(sam_schema)

# Inject CloudFormation documentation to SAM pass-through properties
def replace_passthrough(d: Dict[str, Any]) -> Dict[str, Any]:
def replace_passthrough(d: dict[str, Any]) -> dict[str, Any]:
passthrough = d["__samPassThrough"]
schema = deepcopy(_deep_get(cfn_schema, passthrough["schemaPath"]))
schema["markdownDescription"] = passthrough["markdownDescriptionOverride"]
3 changes: 1 addition & 2 deletions setup.py
Original file line number Diff line number Diff line change
@@ -1,4 +1,3 @@
# -*- coding: utf-8 -*-
#
# setup.py
#
@@ -32,7 +31,7 @@ def read(*filenames, **kwargs):
sep = kwargs.get("sep", os.linesep)
buf = []
for filename in filenames:
with io.open(filename, encoding=encoding) as f:
with open(filename, encoding=encoding) as f:
buf.append(f.read())
return sep.join(buf)

2 changes: 1 addition & 1 deletion tests/policy_template_processor/test_processor.py
Original file line number Diff line number Diff line change
@@ -50,7 +50,7 @@ def test_init_must_convert_template_value_dict_to_object(
processor = PolicyTemplatesProcessor(policy_templates_dict)

self.assertEqual(2, len(processor.policy_templates))
self.assertEqual(set(["key1", "key2"]), set(processor.policy_templates.keys()))
self.assertEqual({"key1", "key2"}, set(processor.policy_templates.keys()))

# Template.from_dict must be called only once for each template entry
self.assertEqual(2, template_from_dict_mock.call_count)
2 changes: 1 addition & 1 deletion tests/utils/test_py27hash_fix.py
Original file line number Diff line number Diff line change
@@ -497,7 +497,7 @@ def test_with_list(self):
self.assertIsInstance(item, Py27Dict)

def test_with_other_type(self):
original = [("a", "b"), set(["a", "b"]), 123, 123.123]
original = [("a", "b"), {"a", "b"}, 123, 123.123]
converted = _convert_to_py27_type(original)
self.assertIsInstance(converted[0], tuple)
self.assertIsInstance(converted[1], set)