def set_model_data()

in pai/model/_model.py [0:0]

• 44 lines of code
• 12 McCabe index (conditional complexity)

    def set_model_data(self, model_data: str, mount_path: Optional[str] = None):
        """
        Set the model data for the InferenceSpec instance.

        Args:
            model_data (str): The model data to be set. It must be an OSS URI.
            mount_path (str, optional): The mount path in the container.

        Raises:
            DuplicatedMountException: If the model data is already mounted to the container.
        """

        def is_model_storage(storage: Dict[str, Any]):
            return (
                "properties" in storage
                and storage["properties"].get("resource_type") == "model"
            )

        if not model_data:
            return
        if not self.is_container_serving():
            # if model_data is an OSS URI with endpoint, truncate the endpoint.
            oss_uri_obj = OssUriObj(model_data)
            model_path_uri = "oss://{bucket_name}/{key}".format(
                bucket_name=oss_uri_obj.bucket_name,
                key=oss_uri_obj.object_key,
            )
            self.add_option("model_path", model_path_uri)
        else:
            indexes = [idx for idx, s in enumerate(self.storage) if is_model_storage(s)]
            # replace the first model storage with the model_data.
            if indexes:
                if len(indexes) > 1:
                    logger.warning(
                        "Multiple model storage found in the InferenceSpec,"
                        " use the first one."
                    )
                idx = indexes[0]
                oss_uri_obj = OssUriObj(model_data)

                storage_config = {
                    "path": oss_uri_obj.get_dir_uri(),
                }

                if oss_uri_obj.endpoint:
                    storage_config.update(
                        {
                            "endpoint": oss_uri_obj.endpoint,
                        }
                    )
                self.storage[idx].oss = self._transform_value(storage_config)
            else:
                try:
                    self.mount(
                        model_data,
                        mount_path=mount_path or DefaultServiceConfig.model_path,
                        properties={"resource_type": "model", "resource_use": "base"},
                    )
                except DuplicatedMountException as e:
                    # ignore duplicated mount
                    logger.warning("Model is already mounted the container: %s", e)