in src/containerapp/azext_containerapp/containerapp_decorator.py [0:0]
def set_up_create_containerapp_yaml(self, name, file_name):
# This list can not include anything that has a default value: transport, revisions_mode, environment_type
if self.get_argument_image() or self.get_argument_min_replicas() or self.get_argument_max_replicas() or self.get_argument_target_port() or self.get_argument_ingress() or \
self.get_argument_secrets() or self.get_argument_env_vars() or self.get_argument_cpu() or self.get_argument_memory() or self.get_argument_registry_server() or \
self.get_argument_registry_user() or self.get_argument_registry_pass() or self.get_argument_dapr_enabled() or self.get_argument_dapr_app_port() or self.get_argument_dapr_app_id() or \
self.get_argument_startup_command() or self.get_argument_args() or self.get_argument_tags() or self.get_argument_target_label():
not self.get_argument_disable_warnings() and logger.warning(
'Additional flags were passed along with --yaml. These flags will be ignored, and the configuration defined in the yaml will be used instead')
yaml_containerapp = process_loaded_yaml(load_yaml_file(file_name))
if not yaml_containerapp.get('name'):
yaml_containerapp['name'] = name
elif yaml_containerapp.get('name').lower() != name.lower():
logger.warning(
'The app name provided in the --yaml file "{}" does not match the one provided in the --name flag "{}". The one provided in the --yaml file will be used.'.format(
yaml_containerapp.get('name'), name))
name = yaml_containerapp.get('name')
if not yaml_containerapp.get('type'):
yaml_containerapp['type'] = 'Microsoft.App/containerApps'
elif yaml_containerapp.get('type').lower() != "microsoft.app/containerapps":
raise ValidationError('Containerapp type must be \"Microsoft.App/ContainerApps\"')
# Deserialize the yaml into a ContainerApp object. Need this since we're not using SDK
try:
deserializer = create_deserializer(self.models)
self.containerapp_def = deserializer('ContainerApp', yaml_containerapp)
except DeserializationError as ex:
raise ValidationError(
'Invalid YAML provided. Please see https://aka.ms/azure-container-apps-yaml for a valid containerapps YAML spec.') from ex
# Remove tags before converting from snake case to camel case, then re-add tags. We don't want to change the case of the tags. Need this since we're not using SDK
tags = None
if yaml_containerapp.get('tags'):
tags = yaml_containerapp.get('tags')
del yaml_containerapp['tags']
# Save customizedKeys before converting from snake case to camel case, then re-add customizedKeys. We don't want to change the case of the customizedKeys.
service_binds = safe_get(yaml_containerapp, "properties", "template", "serviceBinds", default=[])
customized_keys_dict = {}
for bind in service_binds:
if bind.get("name") and bind.get("customizedKeys"):
customized_keys_dict[bind["name"]] = bind["customizedKeys"]
self.containerapp_def = _convert_object_from_snake_to_camel_case(_object_to_dict(self.containerapp_def))
self.containerapp_def['tags'] = tags
# Containerapp object is deserialized from 'ContainerApp' model, "Properties" level is lost after deserialization
# We try to get serviceBinds from 'properties.template.serviceBinds' first, if not exists, try 'template.serviceBinds'
service_binds = safe_get(self.containerapp_def, "properties", "template", "serviceBinds") or safe_get(self.containerapp_def, "template", "serviceBinds")
if service_binds:
for bind in service_binds:
if bind.get("name") and bind.get("customizedKeys"):
bind["customizedKeys"] = customized_keys_dict.get(bind["name"])
# After deserializing, some properties may need to be moved under the "properties" attribute. Need this since we're not using SDK
self.containerapp_def = process_loaded_yaml(self.containerapp_def)
# Remove "additionalProperties" and read-only attributes that are introduced in the deserialization. Need this since we're not using SDK
_remove_additional_attributes(self.containerapp_def)
_remove_readonly_attributes(self.containerapp_def)
# Remove extra workloadProfileName introduced in deserialization
if "workloadProfileName" in self.containerapp_def:
del self.containerapp_def["workloadProfileName"]
# Validate managed environment
env_id = self.containerapp_def["properties"]['environmentId']
env_info = None
if self.get_argument_managed_env():
if not self.get_argument_disable_warnings() and env_id is not None and env_id != self.get_argument_managed_env():
logger.warning('The environmentId was passed along with --yaml. The value entered with --environment will be ignored, and the configuration defined in the yaml will be used instead')
if env_id is None:
env_id = self.get_argument_managed_env()
safe_set(self.containerapp_def, "properties", "environmentId", value=env_id)
if not self.containerapp_def["properties"].get('environmentId'):
raise RequiredArgumentMissingError(
'environmentId is required. This can be retrieved using the `az containerapp env show -g MyResourceGroup -n MyContainerappEnvironment --query id` command. Please see https://aka.ms/azure-container-apps-yaml for a valid containerapps YAML spec.')
if is_valid_resource_id(env_id):
parsed_managed_env = parse_resource_id(env_id)
env_name = parsed_managed_env['name']
env_rg = parsed_managed_env['resource_group']
else:
raise ValidationError('Invalid environmentId specified. Environment not found')
try:
env_info = self.get_environment_client().show(cmd=self.cmd, resource_group_name=env_rg, name=env_name)
except Exception as e:
handle_non_404_status_code_exception(e)
if not env_info:
raise ValidationError("The environment '{}' in resource group '{}' was not found".format(env_name, env_rg))
# Validate location
if not self.containerapp_def.get('location'):
self.containerapp_def['location'] = env_info['location']