# --------------------------------------------------------- # Copyright (c) Microsoft Corporation. All rights reserved. # --------------------------------------------------------- # pylint: disable=protected-access, too-many-instance-attributes import os from pathlib import Path from typing import Any, Dict, Optional, Union import yaml # type: ignore[import] from azure.ai.ml._exception_helper import log_and_raise_error from azure.ai.ml._restclient.v2023_04_01_preview.models import BuildContext as RestBuildContext from azure.ai.ml._restclient.v2023_04_01_preview.models import ( EnvironmentContainer, EnvironmentVersion, EnvironmentVersionProperties, ) from azure.ai.ml._schema import EnvironmentSchema from azure.ai.ml._utils._arm_id_utils import AMLVersionedArmId from azure.ai.ml._utils._asset_utils import get_ignore_file, get_object_hash from azure.ai.ml._utils.utils import dump_yaml, is_url, load_file, load_yaml from azure.ai.ml.constants._common import ANONYMOUS_ENV_NAME, BASE_PATH_CONTEXT_KEY, PARAMS_OVERRIDE_KEY, ArmConstants from azure.ai.ml.entities._assets.asset import Asset from azure.ai.ml.entities._assets.intellectual_property import IntellectualProperty from azure.ai.ml.entities._mixins import LocalizableMixin from azure.ai.ml.entities._system_data import SystemData from azure.ai.ml.entities._util import get_md5_string, load_from_dict from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationErrorType, ValidationException class BuildContext: """Docker build context for Environment. :param path: The local or remote path to the the docker build context directory. :type path: Union[str, os.PathLike] :param dockerfile_path: The path to the dockerfile relative to root of docker build context directory. :type dockerfile_path: str .. admonition:: Example: .. literalinclude:: ../samples/ml_samples_misc.py :start-after: [START build_context_entity_create] :end-before: [END build_context_entity_create] :language: python :dedent: 8 :caption: Create a Build Context object. """ def __init__( self, *, dockerfile_path: Optional[str] = None, path: Optional[Union[str, os.PathLike]] = None, ): self.dockerfile_path = dockerfile_path self.path = path def _to_rest_object(self) -> RestBuildContext: return RestBuildContext(context_uri=self.path, dockerfile_path=self.dockerfile_path) @classmethod def _from_rest_object(cls, rest_obj: RestBuildContext) -> "BuildContext": return BuildContext( path=rest_obj.context_uri, dockerfile_path=rest_obj.dockerfile_path, ) def __eq__(self, other: Any) -> bool: res: bool = self.dockerfile_path == other.dockerfile_path and self.path == other.path return res def __ne__(self, other: Any) -> bool: return not self.__eq__(other) class Environment(Asset, LocalizableMixin): """Environment for training. :param name: Name of the resource. :type name: str :param version: Version of the asset. :type version: str :param description: Description of the resource. :type description: str :param image: URI of a custom base image. :type image: str :param build: Docker build context to create the environment. Mutually exclusive with "image" :type build: ~azure.ai.ml.entities._assets.environment.BuildContext :param conda_file: Path to configuration file listing conda packages to install. :type conda_file: typing.Union[str, os.PathLike] :param tags: Tag dictionary. Tags can be added, removed, and updated. :type tags: dict[str, str] :param properties: The asset property dictionary. :type properties: dict[str, str] :param datastore: The datastore to upload the local artifact to. :type datastore: str :param kwargs: A dictionary of additional configuration parameters. :type kwargs: dict .. admonition:: Example: .. literalinclude:: ../samples/ml_samples_misc.py :start-after: [START env_entity_create] :end-before: [END env_entity_create] :language: python :dedent: 8 :caption: Create a Environment object. """ def __init__( self, *, name: Optional[str] = None, version: Optional[str] = None, description: Optional[str] = None, image: Optional[str] = None, build: Optional[BuildContext] = None, conda_file: Optional[Union[str, os.PathLike, Dict]] = None, tags: Optional[Dict] = None, properties: Optional[Dict] = None, datastore: Optional[str] = None, **kwargs: Any, ): self._arm_type: str = "" self.latest_version: str = "" # type: ignore[assignment] self.image: Optional[str] = None inference_config = kwargs.pop("inference_config", None) os_type = kwargs.pop("os_type", None) self._intellectual_property = kwargs.pop("intellectual_property", None) super().__init__( name=name, version=version, description=description, tags=tags, properties=properties, **kwargs, ) self.conda_file = conda_file self.image = image self.build = build self.inference_config = inference_config self.os_type = os_type self._arm_type = ArmConstants.ENVIRONMENT_VERSION_TYPE self._conda_file_path = ( _resolve_path(base_path=self.base_path, input=conda_file) if isinstance(conda_file, (os.PathLike, str)) else None ) self.path = None self.datastore = datastore self._upload_hash = None self._translated_conda_file = None if self.conda_file: self._translated_conda_file = dump_yaml(self.conda_file, sort_keys=True) # service needs str representation if self.build and self.build.path and not is_url(self.build.path): path = Path(self.build.path) if not path.is_absolute(): path = Path(self.base_path, path).resolve() self.path = path if self._is_anonymous: if self.path: self._ignore_file = get_ignore_file(path) self._upload_hash = get_object_hash(path, self._ignore_file) self._generate_anonymous_name_version(source="build") elif self.image: self._generate_anonymous_name_version( source="image", conda_file=self._translated_conda_file, inference_config=self.inference_config ) @property def conda_file(self) -> Optional[Union[str, os.PathLike, Dict]]: """Conda environment specification. :return: Conda dependencies loaded from `conda_file` param. :rtype: Optional[Union[str, os.PathLike]] """ return self._conda_file @conda_file.setter def conda_file(self, value: Optional[Union[str, os.PathLike, Dict]]) -> None: """Set conda environment specification. :param value: A path to a local conda dependencies yaml file or a loaded yaml dictionary of dependencies. :type value: Union[str, os.PathLike, Dict] :return: None """ if not isinstance(value, Dict): value = _deserialize(self.base_path, value, is_conda=True) self._conda_file = value @classmethod def _load( cls, data: Optional[dict] = None, yaml_path: Optional[Union[os.PathLike, str]] = None, params_override: Optional[list] = None, **kwargs: Any, ) -> "Environment": params_override = params_override or [] data = data or {} context = { BASE_PATH_CONTEXT_KEY: Path(yaml_path).parent if yaml_path else Path("./"), PARAMS_OVERRIDE_KEY: params_override, } res: Environment = load_from_dict(EnvironmentSchema, data, context, **kwargs) return res def _to_rest_object(self) -> EnvironmentVersion: self.validate() environment_version = EnvironmentVersionProperties() if self.conda_file: environment_version.conda_file = self._translated_conda_file if self.image: environment_version.image = self.image if self.build: environment_version.build = self.build._to_rest_object() if self.os_type: environment_version.os_type = self.os_type if self.tags: environment_version.tags = self.tags if self._is_anonymous: environment_version.is_anonymous = self._is_anonymous if self.inference_config: environment_version.inference_config = self.inference_config if self.description: environment_version.description = self.description if self.properties: environment_version.properties = self.properties environment_version_resource = EnvironmentVersion(properties=environment_version) return environment_version_resource @classmethod def _from_rest_object(cls, env_rest_object: EnvironmentVersion) -> "Environment": rest_env_version = env_rest_object.properties arm_id = AMLVersionedArmId(arm_id=env_rest_object.id) environment = Environment( id=env_rest_object.id, name=arm_id.asset_name, version=arm_id.asset_version, description=rest_env_version.description, tags=rest_env_version.tags, creation_context=( SystemData._from_rest_object(env_rest_object.system_data) if env_rest_object.system_data else None ), is_anonymous=rest_env_version.is_anonymous, image=rest_env_version.image, os_type=rest_env_version.os_type, inference_config=rest_env_version.inference_config, build=BuildContext._from_rest_object(rest_env_version.build) if rest_env_version.build else None, properties=rest_env_version.properties, intellectual_property=( IntellectualProperty._from_rest_object(rest_env_version.intellectual_property) if rest_env_version.intellectual_property else None ), ) if rest_env_version.conda_file: translated_conda_file = yaml.safe_load(rest_env_version.conda_file) environment.conda_file = translated_conda_file environment._translated_conda_file = rest_env_version.conda_file return environment @classmethod def _from_container_rest_object(cls, env_container_rest_object: EnvironmentContainer) -> "Environment": env = Environment( name=env_container_rest_object.name, version="1", id=env_container_rest_object.id, creation_context=SystemData._from_rest_object(env_container_rest_object.system_data), ) env.latest_version = env_container_rest_object.properties.latest_version # Setting version to None since if version is not provided it is defaulted to "1". # This should go away once container concept is finalized. env.version = None return env def _to_arm_resource_param(self, **kwargs: Any) -> Dict: # pylint: disable=unused-argument properties = self._to_rest_object().properties return { self._arm_type: { ArmConstants.NAME: self.name, ArmConstants.VERSION: self.version, ArmConstants.PROPERTIES_PARAMETER_NAME: self._serialize.body(properties, "EnvironmentVersion"), } } def _to_dict(self) -> Dict: res: dict = EnvironmentSchema(context={BASE_PATH_CONTEXT_KEY: "./"}).dump(self) return res def validate(self) -> None: """Validate the environment by checking its name, image and build .. admonition:: Example: .. literalinclude:: ../samples/ml_samples_misc.py :start-after: [START env_entities_validate] :end-before: [END env_entities_validate] :language: python :dedent: 8 :caption: Validate environment example. """ if self.name is None: msg = "Environment name is required" err = ValidationException( message=msg, target=ErrorTarget.ENVIRONMENT, no_personal_data_message=msg, error_category=ErrorCategory.USER_ERROR, error_type=ValidationErrorType.MISSING_FIELD, ) log_and_raise_error(err) if self.image is None and self.build is None: msg = "Docker image or Dockerfile is required for environments" err = ValidationException( message=msg, target=ErrorTarget.ENVIRONMENT, no_personal_data_message=msg, error_category=ErrorCategory.USER_ERROR, error_type=ValidationErrorType.MISSING_FIELD, ) log_and_raise_error(err) if self.image and self.build: msg = "Docker image or Dockerfile should be provided not both" err = ValidationException( message=msg, target=ErrorTarget.ENVIRONMENT, no_personal_data_message=msg, error_category=ErrorCategory.USER_ERROR, error_type=ValidationErrorType.INVALID_VALUE, ) log_and_raise_error(err) def __eq__(self, other: object) -> bool: if not isinstance(other, Environment): return NotImplemented return ( self.name == other.name and self.id == other.id and self.version == other.version and self.description == other.description and self.tags == other.tags and self.properties == other.properties and self.base_path == other.base_path and self.image == other.image and self.build == other.build and self.conda_file == other.conda_file and self.inference_config == other.inference_config and self._is_anonymous == other._is_anonymous and self.os_type == other.os_type and self._intellectual_property == other._intellectual_property ) def __ne__(self, other: object) -> bool: return not self.__eq__(other) def _generate_anonymous_name_version( self, source: str, conda_file: Optional[str] = None, inference_config: Optional[Dict] = None ) -> None: hash_str = "" if source == "image": hash_str = hash_str.join(get_md5_string(self.image)) if inference_config: hash_str = hash_str.join(get_md5_string(yaml.dump(inference_config, sort_keys=True))) if conda_file: hash_str = hash_str.join(get_md5_string(conda_file)) if source == "build": if self.build is not None and not self.build.dockerfile_path: hash_str = hash_str.join(get_md5_string(self._upload_hash)) else: if self.build is not None: hash_str = hash_str.join(get_md5_string(self._upload_hash)).join( get_md5_string(self.build.dockerfile_path) ) version_hash = get_md5_string(hash_str) self.version = version_hash self.name = ANONYMOUS_ENV_NAME def _localize(self, base_path: str) -> None: """Called on an asset got from service to clean up remote attributes like id, creation_context, etc. and update base_path. :param base_path: The base path :type base_path: str """ if not getattr(self, "id", None): raise ValueError("Only remote asset can be localize but got a {} without id.".format(type(self))) self._id = None self._creation_context = None self._base_path = base_path if self._is_anonymous: self.name, self.version = None, None # TODO: Remove _DockerBuild and _DockerConfiguration classes once local endpoint moves to using updated env class _DockerBuild: """Helper class to encapsulate Docker build info for Environment.""" def __init__( self, base_path: Optional[Union[str, os.PathLike]] = None, dockerfile: Optional[str] = None, ): self.dockerfile = _deserialize(base_path, dockerfile) @classmethod def _to_rest_object(cls) -> None: return None def _from_rest_object(self, rest_obj: Any) -> None: self.dockerfile = rest_obj.dockerfile def __eq__(self, other: Any) -> bool: res: bool = self.dockerfile == other.dockerfile return res def __ne__(self, other: Any) -> bool: return not self.__eq__(other) def _deserialize( base_path: Optional[Union[str, os.PathLike]], input: Optional[Union[str, os.PathLike, Dict]], # pylint: disable=redefined-builtin is_conda: bool = False, ) -> Optional[Union[str, os.PathLike, Dict]]: """Deserialize user input files for conda and docker. :param base_path: The base path for all files supplied by user. :type base_path: Union[str, os.PathLike] :param input: Input to be deserialized. Will be either dictionary of file contents or path to file. :type input: Union[str, os.PathLike, Dict[str, str]] :param is_conda: If file is conda file, it will be returned as dictionary :type is_conda: bool :return: The deserialized data :rtype: Union[str, Dict] """ if input: path = _resolve_path(base_path=base_path, input=input) data: Union[str, Dict] = "" if is_conda: data = load_yaml(path) else: data = load_file(path) return data return input def _resolve_path(base_path: Any, input: Any) -> Path: # pylint: disable=redefined-builtin """Deserialize user input files for conda and docker. :param base_path: The base path for all files supplied by user. :type base_path: Union[str, os.PathLike] :param input: Input to be deserialized. Will be either dictionary of file contents or path to file. :type input: Union[str, os.PathLike, Dict[str, str]] :return: The resolved path :rtype: Path """ path = Path(input) if not path.is_absolute(): path = Path(base_path, path).resolve() return path