in samtranslator/model/stepfunctions/events.py [0:0]
def _add_swagger_integration(self, api, resource, role, intrinsics_resolver):
"""Adds the path and method for this Api event source to the Swagger body for the provided RestApi.
:param model.apigateway.ApiGatewayRestApi rest_api: the RestApi to which the path and method should be added.
"""
swagger_body = api.get("DefinitionBody")
if swagger_body is None:
return
resource_arn = resource.get_runtime_attr("arn")
integration_uri = fnSub("arn:${AWS::Partition}:apigateway:${AWS::Region}:states:action/StartExecution")
editor = SwaggerEditor(swagger_body)
if editor.has_integration(self.Path, self.Method):
# Cannot add the integration, if it is already present
raise InvalidEventException(
self.relative_id,
'API method "{method}" defined multiple times for path "{path}".'.format(
method=self.Method, path=self.Path
),
)
condition = None
if CONDITION in resource.resource_attributes:
condition = resource.resource_attributes[CONDITION]
editor.add_state_machine_integration(
self.Path,
self.Method,
integration_uri,
role.get_runtime_attr("arn"),
self._generate_request_template(resource),
condition=condition,
)
# Note: Refactor and combine the section below with the Api eventsource for functions
if self.Auth:
method_authorizer = self.Auth.get("Authorizer")
api_auth = api.get("Auth")
api_auth = intrinsics_resolver.resolve_parameter_refs(api_auth)
if method_authorizer:
api_authorizers = api_auth and api_auth.get("Authorizers")
if method_authorizer != "AWS_IAM":
if method_authorizer != "NONE" and not api_authorizers:
raise InvalidEventException(
self.relative_id,
"Unable to set Authorizer [{authorizer}] on API method [{method}] for path [{path}] "
"because the related API does not define any Authorizers.".format(
authorizer=method_authorizer, method=self.Method, path=self.Path
),
)
if method_authorizer != "NONE" and not api_authorizers.get(method_authorizer):
raise InvalidEventException(
self.relative_id,
"Unable to set Authorizer [{authorizer}] on API method [{method}] for path [{path}] "
"because it wasn't defined in the API's Authorizers.".format(
authorizer=method_authorizer, method=self.Method, path=self.Path
),
)
if method_authorizer == "NONE":
if not api_auth or not api_auth.get("DefaultAuthorizer"):
raise InvalidEventException(
self.relative_id,
"Unable to set Authorizer on API method [{method}] for path [{path}] because 'NONE' "
"is only a valid value when a DefaultAuthorizer on the API is specified.".format(
method=self.Method, path=self.Path
),
)
if self.Auth.get("AuthorizationScopes") and not isinstance(self.Auth.get("AuthorizationScopes"), list):
raise InvalidEventException(
self.relative_id,
"Unable to set Authorizer on API method [{method}] for path [{path}] because "
"'AuthorizationScopes' must be a list of strings.".format(method=self.Method, path=self.Path),
)
apikey_required_setting = self.Auth.get("ApiKeyRequired")
apikey_required_setting_is_false = apikey_required_setting is not None and not apikey_required_setting
if apikey_required_setting_is_false and (not api_auth or not api_auth.get("ApiKeyRequired")):
raise InvalidEventException(
self.relative_id,
"Unable to set ApiKeyRequired [False] on API method [{method}] for path [{path}] "
"because the related API does not specify any ApiKeyRequired.".format(
method=self.Method, path=self.Path
),
)
if method_authorizer or apikey_required_setting is not None:
editor.add_auth_to_method(api=api, path=self.Path, method_name=self.Method, auth=self.Auth)
if self.Auth.get("ResourcePolicy"):
resource_policy = self.Auth.get("ResourcePolicy")
editor.add_resource_policy(resource_policy=resource_policy, path=self.Path, stage=self.Stage)
if resource_policy.get("CustomStatements"):
editor.add_custom_statements(resource_policy.get("CustomStatements"))
api["DefinitionBody"] = editor.swagger