in samtranslator/model/eventsources/push.py [0:0]
def _add_swagger_integration(self, api, function, 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
partition = ArnGenerator.get_partition_name()
uri = _build_apigw_integration_uri(function, partition)
editor = SwaggerEditor(swagger_body)
if editor.has_integration(self.Path, self.Method):
# Cannot add the Lambda 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 function.resource_attributes:
condition = function.resource_attributes[CONDITION]
editor.add_lambda_integration(self.Path, self.Method, uri, self.Auth, api.get("Auth"), condition=condition)
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 not isinstance(method_authorizer, str):
raise InvalidEventException(
self.relative_id,
"Unable to set Authorizer [{authorizer}] on API method [{method}] for path [{path}] "
"because it wasn't defined with acceptable values in the API's 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"))
if self.RequestModel:
method_model = self.RequestModel.get("Model")
if method_model:
api_models = api.get("Models")
if not api_models:
raise InvalidEventException(
self.relative_id,
"Unable to set RequestModel [{model}] on API method [{method}] for path [{path}] "
"because the related API does not define any Models.".format(
model=method_model, method=self.Method, path=self.Path
),
)
if not isinstance(method_model, str):
raise InvalidEventException(
self.relative_id,
"Unable to set RequestModel [{model}] on API method [{method}] for path [{path}] "
"because the related API does not contain valid Models.".format(
model=method_model, method=self.Method, path=self.Path
),
)
if not api_models.get(method_model):
raise InvalidEventException(
self.relative_id,
"Unable to set RequestModel [{model}] on API method [{method}] for path [{path}] "
"because it wasn't defined in the API's Models.".format(
model=method_model, method=self.Method, path=self.Path
),
)
editor.add_request_model_to_method(
path=self.Path, method_name=self.Method, request_model=self.RequestModel
)
validate_body = self.RequestModel.get("ValidateBody")
validate_parameters = self.RequestModel.get("ValidateParameters")
# Checking if any of the fields are defined as it can be false we are checking if the field are not None
if validate_body is not None or validate_parameters is not None:
# as we are setting two different fields we are here setting as default False
# In case one of them are not defined
validate_body = False if validate_body is None else validate_body
validate_parameters = False if validate_parameters is None else validate_parameters
# If not type None but any other type it should explicitly invalidate the Spec
# Those fields should be only a boolean
if not isinstance(validate_body, bool) or not isinstance(validate_parameters, bool):
raise InvalidEventException(
self.relative_id,
"Unable to set Validator to RequestModel [{model}] on API method [{method}] for path [{path}] "
"ValidateBody and ValidateParameters must be a boolean type, strings or intrinsics are not supported.".format(
model=method_model, method=self.Method, path=self.Path
),
)
editor.add_request_validator_to_method(
path=self.Path,
method_name=self.Method,
validate_body=validate_body,
validate_parameters=validate_parameters,
)
if self.RequestParameters:
default_value = {"Required": False, "Caching": False}
parameters = []
for parameter in self.RequestParameters:
if isinstance(parameter, dict):
parameter_name, parameter_value = next(iter(parameter.items()))
if not re.match(r"method\.request\.(querystring|path|header)\.", parameter_name):
raise InvalidEventException(
self.relative_id,
"Invalid value for 'RequestParameters' property. Keys must be in the format "
"'method.request.[querystring|path|header].{value}', "
"e.g 'method.request.header.Authorization'.",
)
if not isinstance(parameter_value, dict) or not all(
key in REQUEST_PARAMETER_PROPERTIES for key in parameter_value.keys()
):
raise InvalidEventException(
self.relative_id,
"Invalid value for 'RequestParameters' property. Values must be an object, "
"e.g { Required: true, Caching: false }",
)
settings = default_value.copy()
settings.update(parameter_value)
settings.update({"Name": parameter_name})
parameters.append(settings)
elif isinstance(parameter, str):
if not re.match(r"method\.request\.(querystring|path|header)\.", parameter):
raise InvalidEventException(
self.relative_id,
"Invalid value for 'RequestParameters' property. Keys must be in the format "
"'method.request.[querystring|path|header].{value}', "
"e.g 'method.request.header.Authorization'.",
)
settings = default_value.copy()
settings.update({"Name": parameter})
parameters.append(settings)
else:
raise InvalidEventException(
self.relative_id,
"Invalid value for 'RequestParameters' property. Property must be either a string or an object",
)
editor.add_request_parameters_to_method(
path=self.Path, method_name=self.Method, request_parameters=parameters
)
api["DefinitionBody"] = editor.swagger