def set_up_container()

in src/containerapp/azext_containerapp/containerapp_job_decorator.py [0:0]

• 83 lines of code
• 43 McCabe index (conditional complexity)

    def set_up_container(self):
        if self.should_update_container():  # pylint: disable=too-many-nested-blocks
            safe_set(self.new_containerappjob, "properties", "template", "containers", value=self.containerappjob_def["properties"]["template"]["containers"])

            if not self.get_argument_container_name():
                if len(self.new_containerappjob["properties"]["template"]["containers"]) == 1:
                    container_name = self.new_containerappjob["properties"]["template"]["containers"][0]["name"]
                    self.set_argument_container_name(container_name)
                else:
                    raise ValidationError(
                        "Usage error: --container-name is required when adding or updating a container")

            # Check if updating existing container
            updating_existing_container = False
            for c in self.new_containerappjob["properties"]["template"]["containers"]:
                if c["name"].lower() == self.get_argument_container_name().lower():
                    updating_existing_container = True

                    if self.get_argument_image() is not None:
                        c["image"] = self.get_argument_image()

                    if self.get_argument_set_env_vars() is not None:
                        if "env" not in c or not c["env"]:
                            c["env"] = []
                        # env vars
                        _add_or_update_env_vars(c["env"], parse_env_var_flags(self.get_argument_set_env_vars()))

                    if self.get_argument_replace_env_vars() is not None:
                        # Remove other existing env_vars, then add them
                        c["env"] = []
                        _add_or_update_env_vars(c["env"], parse_env_var_flags(self.get_argument_replace_env_vars()))

                    if self.get_argument_remove_env_vars() is not None:
                        if "env" not in c or not c["env"]:
                            c["env"] = []
                        # env vars
                        _remove_env_vars(c["env"], self.get_argument_remove_env_vars())

                    if self.get_argument_remove_all_env_vars():
                        c["env"] = []

                    if self.get_argument_startup_command() is not None:
                        if isinstance(self.get_argument_startup_command(), list) and not self.get_argument_startup_command():
                            c["command"] = None
                        else:
                            c["command"] = self.get_argument_startup_command()
                    if self.get_argument_args() is not None:
                        if isinstance(self.get_argument_args(), list) and not self.get_argument_args():
                            c["args"] = None
                        else:
                            c["args"] = self.get_argument_args()
                    if self.get_argument_cpu() is not None or self.get_argument_memory() is not None:
                        if "resources" in c and c["resources"]:
                            if self.get_argument_cpu() is not None:
                                c["resources"]["cpu"] = self.get_argument_cpu()
                            if self.get_argument_memory() is not None:
                                c["resources"]["memory"] = self.get_argument_memory()
                        else:
                            c["resources"] = {
                                "cpu": self.get_argument_cpu(),
                                "memory": self.get_argument_memory()
                            }

            # If not updating existing container, add as new container
            if not updating_existing_container:
                if self.get_argument_image() is None:
                    raise ValidationError("Usage error: --image is required when adding a new container")

                resources_def = None
                if self.get_argument_cpu() is not None or self.get_argument_memory() is not None:
                    resources_def = ContainerResourcesModel
                    resources_def["cpu"] = self.get_argument_cpu()
                    resources_def["memory"] = self.get_argument_memory()

                container_def = ContainerModel
                container_def["name"] = self.get_argument_container_name()
                container_def["image"] = self.get_argument_image()
                container_def["env"] = []

                if self.get_argument_set_env_vars() is not None:
                    # env vars
                    _add_or_update_env_vars(container_def["env"], parse_env_var_flags(self.get_argument_set_env_vars()))

                if self.get_argument_replace_env_vars() is not None:
                    # env vars
                    _add_or_update_env_vars(container_def["env"], parse_env_var_flags(self.get_argument_replace_env_vars()))

                if self.get_argument_remove_env_vars() is not None:
                    # env vars
                    _remove_env_vars(container_def["env"], self.get_argument_remove_env_vars())

                if self.get_argument_remove_all_env_vars():
                    container_def["env"] = []

                if self.get_argument_startup_command() is not None:
                    if isinstance(self.get_argument_startup_command(), list) and not self.get_argument_startup_command():
                        container_def["command"] = None
                    else:
                        container_def["command"] = self.get_argument_startup_command()
                if self.get_argument_args() is not None:
                    if isinstance(self.get_argument_args(), list) and not self.get_argument_args():
                        container_def["args"] = None
                    else:
                        container_def["args"] = self.get_argument_args()
                if resources_def is not None:
                    container_def["resources"] = resources_def

                self.new_containerappjob["properties"]["template"]["containers"].append(container_def)