in samcli/lib/sync/flows/rest_api_sync_flow.py [0:0]
def _collect_stages(self) -> Set[str]:
"""
Collect all stages needed to be updated
Returns
----------
Set[str]: The set of stage names to be updated
"""
# Get the stage name associated with the previous deployment and update stage
# Stage needs to be flushed so that new changes will be visible immediately
api_resource = get_resource_by_id(self._stacks, ResourceIdentifier(self._api_identifier))
stage_resources = get_resource_ids_by_type(self._stacks, AWS_APIGATEWAY_STAGE)
deployment_resources = get_resource_ids_by_type(self._stacks, AWS_APIGATEWAY_DEPLOYMENT)
stages = set()
# If it is a SAM resource, get the StageName property
if api_resource:
if api_resource.get("Type") == AWS_SERVERLESS_API:
# The customer defined stage name
stage_name = api_resource.get("Properties", {}).get("StageName")
if stage_name:
stages.add(cast(str, stage_name))
# The stage called "Stage"
if stage_name != "Stage":
response_sta = cast(Dict, self._api_client.get_stages(restApiId=self._api_physical_id))
for item in response_sta.get("item"): # type: ignore
if item.get("stageName") == "Stage":
stages.add("Stage")
# For both SAM and ApiGateway resource, check if any refs from stage resources
for stage_resource in stage_resources:
# RestApiId is a required field in stage
stage_dict = get_resource_by_id(self._stacks, stage_resource)
if not stage_dict:
continue
rest_api_id = stage_dict.get("Properties", {}).get("RestApiId")
dep_id = stage_dict.get("Properties", {}).get("DeploymentId")
# If the stage doesn't have a deployment associated then no need to update
if dep_id is None:
continue
# If the stage's deployment ID is not static and the rest API ID matchs, then update
for deployment_resource in deployment_resources:
if deployment_resource.resource_iac_id == dep_id and rest_api_id == self._api_identifier:
stages.add(cast(str, stage_dict.get("Properties", {}).get("StageName")))
break
return stages