in packages/blueprints/gen-ai-chatbot/static-assets/chatbot-genai-components/backend/python/app/usecases/publication.py [0:0]
def create_bot_publication(user: User, bot_id: str, bot_publish_input: BotPublishInput):
"""Publish an API for the bot."""
# Check existence and permission of the bot
try:
bot = find_private_bot_by_id(user.id, bot_id)
except RecordNotFoundError:
raise RecordNotFoundError(f"Bot {bot_id} is not found.")
if bot.public_bot_id is None:
raise ValueError(f"Bot {bot_id} is not shared. Cannot publish.")
if bot.published_api_codebuild_id is not None:
codebuild_status = find_build_status_by_build_id(bot.published_api_codebuild_id)
if codebuild_status not in ["SUCCEEDED", "FAILED"]:
raise ResourceConflictError(
f"Bot {bot_id} publication is already requested (build id: {bot.published_api_codebuild_id}). Please wait until the previous publication is completed."
)
else:
raise ValueError(
f"Bot {bot_id} is already published. Please remove the publication before re-publishing."
)
# Same value as `bot_id` is used for `public_bot_id`
published_api_id = bot_id
environment_variables = {}
environment_variables["PUBLISHED_API_ID"] = published_api_id
# Set environment variables.
# NOTE: default values are set in `cdk/lib/constructs/api-publish-codebuild.ts`
if bot_publish_input.throttle.rate_limit is not None:
environment_variables["PUBLISHED_API_THROTTLE_RATE_LIMIT"] = str(
bot_publish_input.throttle.rate_limit
)
if bot_publish_input.throttle.burst_limit is not None:
environment_variables["PUBLISHED_API_THROTTLE_BURST_LIMIT"] = str(
bot_publish_input.throttle.burst_limit
)
if bot_publish_input.quota.limit is not None:
environment_variables["PUBLISHED_API_QUOTA_LIMIT"] = str(
bot_publish_input.quota.limit
)
if bot_publish_input.quota.period is not None:
environment_variables["PUBLISHED_API_QUOTA_PERIOD"] = str(
bot_publish_input.quota.period
)
if bot_publish_input.stage is not None:
environment_variables["PUBLISHED_API_DEPLOYMENT_STAGE"] = str(
bot_publish_input.stage
)
if bot_publish_input.allowed_origins is not None:
environment_variables["PUBLISHED_API_ALLOWED_ORIGINS"] = (
str(bot_publish_input.allowed_origins).replace(" ", "").replace("'", '"')
)
# Create `ApiPublishmentStack` by CodeBuild
try:
build_id = start_codebuild_project(environment_variables=environment_variables)
except Exception as e:
raise e
# Update bot attribute
update_bot_publication(
user.id, bot_id, published_api_id=published_api_id, build_id=build_id
)
return