in chalice/deploy/planner.py [0:0]
def _plan_websocketapi(self, resource):
# type: (models.WebsocketAPI) -> Sequence[InstructionMsg]
configs = self._create_websocket_function_configs(resource)
routes = resource.routes
# Which lambda function we use here does not matter. We are only using
# it to find the account id and the region.
lambda_arn_var = list(configs.values())[0]['lambda_arn_var']
shared_plan_preamble = self._arn_parse_instructions(lambda_arn_var) + [
models.JPSearch('dns_suffix',
input_var='parsed_lambda_arn',
output_var='dns_suffix'),
] # type: List[InstructionMsg]
# There's also a set of instructions that are needed
# at the end of deploying a websocket API that apply to both
# the update and create case.
shared_plan_epilogue = [
models.StoreValue(
name='websocket_api_url',
value=StringFormat(
'wss://{websocket_api_id}.execute-api.{region_name}'
'.{dns_suffix}/%s/' % resource.api_gateway_stage,
['websocket_api_id', 'region_name', 'dns_suffix'],
),
),
models.RecordResourceVariable(
resource_type='websocket_api',
resource_name=resource.resource_name,
name='websocket_api_url',
variable_name='websocket_api_url',
),
models.RecordResourceVariable(
resource_type='websocket_api',
resource_name=resource.resource_name,
name='websocket_api_id',
variable_name='websocket_api_id',
),
] # type: List[InstructionMsg]
shared_plan_epilogue += [
models.APICall(
method_name='add_permission_for_apigateway_v2',
params={'function_name': function_config['name'],
'region_name': Variable('region_name'),
'account_id': Variable('account_id'),
'api_id': Variable('websocket_api_id')},
) for function_config in configs.values()
]
main_plan = [] # type: List[InstructionMsg]
if not self._remote_state.resource_exists(resource):
# The resource does not exist, we create it in full here.
main_plan += [
(models.APICall(
method_name='create_websocket_api',
params={'name': resource.name},
output_var='websocket_api_id',
), "Creating websocket api: %s\n" % resource.name),
models.StoreValue(
name='routes',
value=[],
),
]
main_plan += self._inject_websocket_integrations(configs)
for route_key in routes:
main_plan += [self._create_route_for_key(route_key)]
main_plan += [
models.APICall(
method_name='deploy_websocket_api',
params={
'api_id': Variable('websocket_api_id'),
},
output_var='deployment-id',
),
models.APICall(
method_name='create_stage',
params={
'api_id': Variable('websocket_api_id'),
'stage_name': resource.api_gateway_stage,
'deployment_id': Variable('deployment-id'),
}
),
]
else:
# Already exists. Need to sync up the routes, the easiest way to do
# this is to delete them and their integrations and re-create them.
# They will not work if the lambda function changes from under
# them, and the logic for detecting that and making just the needed
# changes is complex. There is an integration test to ensure there
# no dropped messages during a redeployment.
deployed = self._remote_state.resource_deployed_values(resource)
main_plan += [
models.StoreValue(
name='websocket_api_id',
value=deployed['websocket_api_id']
),
models.APICall(
method_name='get_websocket_routes',
params={'api_id': Variable('websocket_api_id')},
output_var='routes',
),
models.APICall(
method_name='delete_websocket_routes',
params={
'api_id': Variable('websocket_api_id'),
'routes': Variable('routes'),
},
),
models.APICall(
method_name='get_websocket_integrations',
params={
'api_id': Variable('websocket_api_id'),
},
output_var='integrations'
),
models.APICall(
method_name='delete_websocket_integrations',
params={
'api_id': Variable('websocket_api_id'),
'integrations': Variable('integrations'),
}
)
]
main_plan += self._inject_websocket_integrations(configs)
for route_key in routes:
main_plan += [self._create_route_for_key(route_key)]
ws_plan = shared_plan_preamble + main_plan + shared_plan_epilogue
if resource.domain_name:
custom_domain_plan = self._add_custom_domain_plan(
resource.domain_name, 'REGIONAL',
)
ws_plan += custom_domain_plan
return ws_plan