in amazon_lex_bot_deploy/amazon_lex_bot_deploy.py [0:0]
def lex_deploy(lex_schema_file=None,
lex_alias=LATEST_ALIAS,
lambda_endpoint=None,
region=None,
log_level='INFO',
example=None):
"""
deploys Amazon Lex schema file to either the $LATEST or a specific alias.
:param lex_schema_file: location of lex schema file. Either example or lex_schema_file have to be set
:param lex_alias: alias to deploy to, default is $LATEST
:param lambda_endpoint: Lambda endpoint ARN. Will replace existing endpoints in Lambda schema
:param region: AWS region to deploy to. default is taken from environment configuration or if nothing is set falls back to 'us-east-1'
:param log_level: DEBUG, INFO, WARNING, ERROR, CRITICAL
:param example: one of "BookTrip", "OrderFlowers", "ScheduleAppointment". Either example or lex_schema_file have to be set
:return: None if success
"""
if example:
SCRIPT_DIR = os.path.dirname(os.path.abspath(__file__))
lex_schema_file = os.path.join(
SCRIPT_DIR, "examples/{}_Export.json".format(example))
logger.setLevel(log_level)
lex_client = setup_boto3_client(region)
with open(lex_schema_file) as lex_schema_file_input:
full_schema = json.load(lex_schema_file_input)
schema_resource = full_schema['resource']
voice_id = schema_resource['voiceId']
bot_ttl = schema_resource['idleSessionTTLInSeconds']
bot_name = schema_resource['name']
child_directed = schema_resource['childDirected']
logger.debug("voice_id: {}".format(voice_id))
# replace existing Lambda endpoints with new one
if lambda_endpoint:
for intent in full_schema['resource']['intents']:
if 'fulfillmentActivity' in intent and 'codeHook' in intent[
'fulfillmentActivity']:
logger.info("changing {} to {}".format(
intent['fulfillmentActivity']['codeHook']['uri'],
lambda_endpoint))
intent['fulfillmentActivity']['codeHook'][
'uri'] = lambda_endpoint
if 'dialogCodeHook' in intent:
logger.info("changing {} to {}".format(
intent['dialogCodeHook']['uri'], lambda_endpoint))
intent['dialogCodeHook']['uri'] = lambda_endpoint
has_lambda_endpoints = True
# check if Lex has permission to call Lambda, if not add the permissions
lambda_endpoints = get_lambda_endpoints(full_schema)
if lambda_endpoints:
create_lambda_permissions(lex_client,
lambda_endpoints,
bot_name=bot_name,
region=region)
buff = io.BytesIO()
zipfile_ob = zipfile.ZipFile(buff, mode='w')
zipfile_ob.writestr(ntpath.basename(lex_schema_file),
json.dumps(full_schema))
buff.seek(0)
try:
start_import_response = lex_client.start_import(
payload=buff.read(),
resourceType='BOT',
mergeStrategy='OVERWRITE_LATEST')
except botocore.exceptions.EndpointConnectionError as ece:
logger.error(ece)
logger.error(
"Maybe Amazon Lex is not supported in region defined. Check https://aws.amazon.com/about-aws/global-infrastructure/regional-product-services/"
)
raise ece
logger.debug("start_import_response: {}".format(start_import_response))
import_id = start_import_response['importId']
wait_async(lex_client.get_import,
'importStatus', ["IN_PROGRESS"], ["FAILED"],
importId=import_id)
bot_intents = []
for intent in schema_resource['intents']:
intent_name = intent['name']
get_intent_response = lex_client.get_intent(name=intent_name,
version=LATEST_ALIAS)
logger.debug("{}, {}".format(intent_name,
get_intent_response['checksum']))
create_intent_version_response = retry_function(
lex_client.create_intent_version,
name=intent_name,
checksum=get_intent_response['checksum'])
bot_intents.append({
'intentName':
intent_name,
'intentVersion':
create_intent_version_response['version']
})
logger.debug(create_intent_version_response)
logger.info("deployed all intents")
get_bot_response = lex_client.get_bot(name=bot_name,
versionOrAlias=LATEST_ALIAS)
logger.debug("STATUS: {}".format(get_bot_response['status']))
wait_async(lex_client.get_bot,
'status', ['BUILDING'], ["FAILED"],
name=bot_name,
versionOrAlias=LATEST_ALIAS)
if "clarificationPrompt" in schema_resource:
put_bot_response = retry_function(
lex_client.put_bot,
name=bot_name,
checksum=get_bot_response['checksum'],
childDirected=child_directed,
locale=schema_resource['locale'],
abortStatement=schema_resource['abortStatement'],
clarificationPrompt=schema_resource['clarificationPrompt'],
intents=bot_intents,
processBehavior='BUILD',
voiceId=voice_id,
idleSessionTTLInSeconds=bot_ttl)
else:
put_bot_response = retry_function(
lex_client.put_bot,
name=bot_name,
checksum=get_bot_response['checksum'],
childDirected=child_directed,
locale=schema_resource['locale'],
abortStatement=schema_resource['abortStatement'],
intents=bot_intents,
processBehavior='BUILD',
voiceId=voice_id,
idleSessionTTLInSeconds=bot_ttl)
logger.debug("put_bot_response: %s", put_bot_response)
response = wait_async(lex_client.get_bot,
'status',
['BUILDING', 'NOT_BUILT', 'READY_BASIC_TESTING'],
["FAILED"],
name=bot_name,
versionOrAlias=LATEST_ALIAS)
create_bot_version_response = retry_function(
lex_client.create_bot_version,
name=bot_name,
checksum=response['checksum'])
new_version = create_bot_version_response['version']
logger.debug("create_bot_version_response: {}".format(
create_bot_version_response))
if lex_alias == LATEST_ALIAS:
logger.debug("deployed to alias: %s, no specific alias given",
lex_alias)
wait_response = wait_async(
lex_client.get_bot,
'status', ['BUILDING', 'NOT_BUILT', 'READY_BASIC_TESTING'],
["FAILED"],
name=bot_name,
versionOrAlias=LATEST_ALIAS)
logger.info("success. bot_status: {} for alias: {}".format(
wait_response['status'], LATEST_ALIAS))
else:
logger.debug("deploying to alias: %s with version: %s.", lex_alias,
new_version)
try:
# check if alias exists, need the checksum if updating
bot_alias = lex_client.get_bot_alias(name=lex_alias,
botName=bot_name)
checksum = bot_alias['checksum']
old_version = bot_alias['botVersion']
if new_version != old_version:
logger.debug(
"new version: {} for existing alias: {} and version: '{}'"
.format(new_version, lex_alias, old_version))
put_bot_alias_response = retry_function(
lex_client.put_bot_alias,
name=lex_alias,
description='latest test',
botVersion=new_version,
botName=bot_name,
checksum=checksum)
logger.debug("put_bot_alias_response : {}".format(
put_bot_alias_response))
# wait for new version
wait_async(lex_client.get_bot,
'version', [old_version],
name=bot_name,
versionOrAlias=lex_alias)
wait_response = wait_async(
lex_client.get_bot,
'status',
['BUILDING', 'NOT_BUILT', 'READY_BASIC_TESTING'],
["FAILED"],
name=bot_name,
versionOrAlias=lex_alias)
logger.info("success. bot_status: {} for alias: {}".format(
wait_response['status'], lex_alias))
else:
logger.info(
'No change for bot. old version == new version ({} == {})'
.format(old_version, new_version))
except lex_client.exceptions.NotFoundException:
# alias not found, create new alias
logger.debug("create new alias: '{}'.".format(lex_alias))
put_bot_alias_response = retry_function(
lex_client.put_bot_alias,
name=lex_alias,
description='latest test',
botVersion=new_version,
botName=bot_name)
logger.debug("put_bot_alias_response : {}".format(
put_bot_alias_response))