diff options
Diffstat (limited to '.venv/lib/python3.12/site-packages/azure/ai/ml/_utils/utils.py')
| -rw-r--r-- | .venv/lib/python3.12/site-packages/azure/ai/ml/_utils/utils.py | 1429 |
1 files changed, 1429 insertions, 0 deletions
diff --git a/.venv/lib/python3.12/site-packages/azure/ai/ml/_utils/utils.py b/.venv/lib/python3.12/site-packages/azure/ai/ml/_utils/utils.py new file mode 100644 index 00000000..74b6352b --- /dev/null +++ b/.venv/lib/python3.12/site-packages/azure/ai/ml/_utils/utils.py @@ -0,0 +1,1429 @@ +# --------------------------------------------------------- +# Copyright (c) Microsoft Corporation. All rights reserved. +# --------------------------------------------------------- + +# pylint: disable=protected-access,too-many-lines +import copy +import decimal +import hashlib +import json +import logging +import os +import random +import re +import string +import sys +import tempfile +import time +import warnings +from collections import OrderedDict +from contextlib import contextmanager, nullcontext +from datetime import timedelta +from functools import singledispatch, wraps +from os import PathLike +from pathlib import Path, PureWindowsPath +from typing import IO, Any, AnyStr, Callable, Dict, Iterable, List, Optional, Tuple, Union +from urllib.parse import urlparse +from uuid import UUID + +import isodate +import pydash +import yaml + +from azure.ai.ml._restclient.v2022_05_01.models import ListViewType, ManagedServiceIdentity +from azure.ai.ml._scope_dependent_operations import OperationScope +from azure.ai.ml._utils._http_utils import HttpPipeline +from azure.ai.ml.constants._common import ( + AZUREML_DISABLE_CONCURRENT_COMPONENT_REGISTRATION, + AZUREML_DISABLE_ON_DISK_CACHE_ENV_VAR, + AZUREML_INTERNAL_COMPONENTS_ENV_VAR, + AZUREML_INTERNAL_COMPONENTS_SCHEMA_PREFIX, + AZUREML_PRIVATE_FEATURES_ENV_VAR, + CommonYamlFields, + DefaultOpenEncoding, + WorkspaceDiscoveryUrlKey, +) +from azure.ai.ml.exceptions import MlException +from azure.core.pipeline.policies import RetryPolicy + +module_logger = logging.getLogger(__name__) + +DEVELOPER_URL_MFE_ENV_VAR = "AZUREML_DEV_URL_MFE" + +# Prefix used when hitting MFE skipping ARM +MFE_PATH_PREFIX = "/mferp/managementfrontend" + + +def _get_mfe_url_override() -> Optional[str]: + return os.getenv(DEVELOPER_URL_MFE_ENV_VAR) + + +def _is_https_url(url: str) -> Union[bool, str]: + if url: + return url.lower().startswith("https") + return False + + +def _csv_parser(text: Optional[str], convert: Callable) -> Optional[str]: + if not text: + return None + if "," in text: + return ",".join(convert(t.strip()) for t in text.split(",")) + + return convert(text) + + +def _snake_to_pascal_convert(text: str) -> str: + return string.capwords(text.replace("_", " ")).replace(" ", "") + + +def snake_to_pascal(text: Optional[str]) -> str: + """Convert snake name to pascal. + + :param text: String to convert + :type text: Optional[str] + :return: + * None if text is None + * Converted text from snake_case to PascalCase + :rtype: Optional[str] + """ + return _csv_parser(text, _snake_to_pascal_convert) + + +def snake_to_kebab(text: Optional[str]) -> Optional[str]: + """Convert snake name to kebab. + + :param text: String to convert + :type text: Optional[str] + :return: + * None if text is None + * Converted text from snake_case to kebab-case + :rtype: Optional[str] + """ + if text: + return re.sub("_", "-", text) + return None + + +# https://stackoverflow.com/questions/1175208 +# This works for pascal to snake as well +def _camel_to_snake_convert(text: str) -> str: + text = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", text) + return re.sub("([a-z0-9])([A-Z])", r"\1_\2", text).lower() + + +def camel_to_snake(text: str) -> Optional[str]: + """Convert camel name to snake. + + :param text: String to convert + :type text: str + :return: + * None if text is None + * Converted text from camelCase to snake_case + :rtype: Optional[str] + """ + return _csv_parser(text, _camel_to_snake_convert) + + +# This is snake to camel back which is different from snake to camel +# https://stackoverflow.com/questions/19053707 +def snake_to_camel(text: Optional[str]) -> Optional[str]: + """Convert snake name to camel. + + :param text: String to convert + :type text: Optional[str] + :return: + * None if text is None + * Converted text from snake_case to camelCase + :rtype: Optional[str] + """ + if text: + return re.sub("_([a-zA-Z0-9])", lambda m: m.group(1).upper(), text) + return None + + +# This is real snake to camel +def _snake_to_camel(name): + return re.sub(r"(?:^|_)([a-z])", lambda x: x.group(1).upper(), name) + + +def float_to_str(f): + """Convert a float to a string without scientific notation. + + :param f: Float to convert + :type f: float + :return: String representation of the float + :rtype: str + """ + with decimal.localcontext() as ctx: + ctx.prec = 20 # Support up to 20 significant figures. + float_as_dec = ctx.create_decimal(repr(f)) + return format(float_as_dec, "f") + + +def create_requests_pipeline_with_retry(*, requests_pipeline: HttpPipeline, retries: int = 3) -> HttpPipeline: + """Creates an HttpPipeline that reuses the same configuration as the supplied pipeline (including the transport), + but overwrites the retry policy. + + :keyword requests_pipeline: Pipeline to base new one off of. + :paramtype requests_pipeline: HttpPipeline + :keyword retries: Number of retries. Defaults to 3. + :paramtype retries: int + :return: Pipeline identical to provided one, except with a new retry policy + :rtype: HttpPipeline + """ + return requests_pipeline.with_policies(retry_policy=get_retry_policy(num_retry=retries)) + + +def get_retry_policy(num_retry: int = 3) -> RetryPolicy: + """Retrieves a retry policy to use in an azure.core.pipeline.Pipeline + + :param num_retry: The number of retries + :type num_retry: int + :return: Returns the msrest or requests REST client retry policy. + :rtype: RetryPolicy + """ + status_forcelist = [413, 429, 500, 502, 503, 504] + backoff_factor = 0.4 + return RetryPolicy( + retry_total=num_retry, + retry_read=num_retry, + retry_connect=num_retry, + retry_backoff_factor=backoff_factor, + retry_on_status_codes=status_forcelist, + ) + + +def download_text_from_url( + source_uri: str, + requests_pipeline: HttpPipeline, + timeout: Optional[Union[float, Tuple[float, float]]] = None, +) -> str: + """Downloads the content from an URL. + + :param source_uri: URI to download + :type source_uri: str + :param requests_pipeline: Used to send the request + :type requests_pipeline: HttpPipeline + :param timeout: One of + * float that specifies the connect and read time outs + * a 2-tuple that specifies the connect and read time out in that order + :type timeout: Union[float, Tuple[float, float]] + :return: The Response text + :rtype: str + """ + if not timeout: + timeout_params = {} + else: + connect_timeout, read_timeout = timeout if isinstance(timeout, tuple) else (timeout, timeout) + timeout_params = {"read_timeout": read_timeout, "connection_timeout": connect_timeout} + + response = requests_pipeline.get(source_uri, **timeout_params) + # Match old behavior from execution service's status API. + if response.status_code == 404: + return "" + + # _raise_request_error(response, "Retrieving content from " + uri) + return response.text() + + +def load_file(file_path: str) -> str: + """Load a local file. + + :param file_path: The relative or absolute path to the local file. + :type file_path: str + :raises ~azure.ai.ml.exceptions.ValidationException: Raised if file or folder cannot be found. + :return: A string representation of the local file's contents. + :rtype: str + """ + from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationErrorType, ValidationException + + # These imports can't be placed in at top file level because it will cause a circular import in + # exceptions.py via _get_mfe_url_override + + try: + with open(file_path, "r", encoding=DefaultOpenEncoding.READ) as f: + cfg = f.read() + except OSError as e: # FileNotFoundError introduced in Python 3 + msg = "No such file or directory: {}" + raise ValidationException( + message=msg.format(file_path), + no_personal_data_message=msg.format("[file_path]"), + error_category=ErrorCategory.USER_ERROR, + target=ErrorTarget.GENERAL, + error_type=ValidationErrorType.FILE_OR_FOLDER_NOT_FOUND, + ) from e + return cfg + + +def load_json(file_path: Optional[Union[str, os.PathLike]]) -> Dict: + """Load a local json file. + + :param file_path: The relative or absolute path to the local file. + :type file_path: Union[str, os.PathLike] + :raises ~azure.ai.ml.exceptions.ValidationException: Raised if file or folder cannot be found. + :return: A dictionary representation of the local file's contents. + :rtype: Dict + """ + from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationErrorType, ValidationException + + # These imports can't be placed in at top file level because it will cause a circular import in + # exceptions.py via _get_mfe_url_override + + try: + with open(file_path, "r", encoding=DefaultOpenEncoding.READ) as f: + cfg = json.load(f) + except OSError as e: # FileNotFoundError introduced in Python 3 + msg = "No such file or directory: {}" + raise ValidationException( + message=msg.format(file_path), + no_personal_data_message=msg.format("[file_path]"), + error_category=ErrorCategory.USER_ERROR, + target=ErrorTarget.GENERAL, + error_type=ValidationErrorType.FILE_OR_FOLDER_NOT_FOUND, + ) from e + return cfg + + +def load_yaml(source: Optional[Union[AnyStr, PathLike, IO]]) -> Dict: + # null check - just return an empty dict. + # Certain CLI commands rely on this behavior to produce a resource + # via CLI, which is then populated through CLArgs. + """Load a local YAML file. + + :param source: Either + * The relative or absolute path to the local file. + * A readable File-like object + :type source: Optional[Union[AnyStr, PathLike, IO]] + :raises ~azure.ai.ml.exceptions.ValidationException: Raised if file or folder cannot be successfully loaded. + Details will be provided in the error message. + :return: A dictionary representation of the local file's contents. + :rtype: Dict + """ + from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationErrorType, ValidationException + + # These imports can't be placed in at top file level because it will cause a circular import in + # exceptions.py via _get_mfe_url_override + + if source is None: + return {} + + if isinstance(source, (str, os.PathLike)): + try: + cm = open(source, "r", encoding=DefaultOpenEncoding.READ) + except OSError as e: + msg = "No such file or directory: {}" + raise ValidationException( + message=msg.format(source), + no_personal_data_message=msg.format("[file_path]"), + error_category=ErrorCategory.USER_ERROR, + target=ErrorTarget.GENERAL, + error_type=ValidationErrorType.FILE_OR_FOLDER_NOT_FOUND, + ) from e + else: + # source is a subclass of IO + if not source.readable(): + msg = "File Permissions Error: The already-open \n\n inputted file is not readable." + raise ValidationException( + message=msg, + no_personal_data_message="File Permissions error", + error_category=ErrorCategory.USER_ERROR, + target=ErrorTarget.GENERAL, + error_type=ValidationErrorType.INVALID_VALUE, + ) + + cm = nullcontext(enter_result=source) + + with cm as f: + try: + return yaml.safe_load(f) + except yaml.YAMLError as e: + msg = f"Error while parsing yaml file: {source} \n\n {str(e)}" + raise ValidationException( + message=msg, + no_personal_data_message="Error while parsing yaml file", + error_category=ErrorCategory.USER_ERROR, + target=ErrorTarget.GENERAL, + error_type=ValidationErrorType.CANNOT_PARSE, + ) from e + + +# pylint: disable-next=docstring-missing-param +def dump_yaml(*args, **kwargs): + """A thin wrapper over yaml.dump which forces `OrderedDict`s to be serialized as mappings. + + Otherwise behaves identically to yaml.dump + + :return: The yaml object + :rtype: Any + """ + + class OrderedDumper(yaml.Dumper): + """A modified yaml serializer that forces pyyaml to represent an OrderedDict as a mapping instead of a + sequence.""" + + OrderedDumper.add_representer(OrderedDict, yaml.representer.SafeRepresenter.represent_dict) + return yaml.dump(*args, Dumper=OrderedDumper, **kwargs) + + +def dump_yaml_to_file( + dest: Optional[Union[AnyStr, PathLike, IO]], + data_dict: Union[OrderedDict, Dict], + default_flow_style=False, + args=None, # pylint: disable=unused-argument + **kwargs, +) -> None: + """Dump dictionary to a local YAML file. + + :param dest: The relative or absolute path where the YAML dictionary will be dumped. + :type dest: Optional[Union[AnyStr, PathLike, IO]] + :param data_dict: Dictionary representing a YAML object + :type data_dict: Union[OrderedDict, Dict] + :param default_flow_style: Use flow style for formatting nested YAML collections + instead of block style. Defaults to False. + :type default_flow_style: bool + :param path: Deprecated. Use 'dest' param instead. + :type path: Optional[Union[AnyStr, PathLike]] + :param args: Deprecated. + :type: Any + :raises ~azure.ai.ml.exceptions.ValidationException: Raised if object cannot be successfully dumped. + Details will be provided in the error message. + """ + from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationErrorType, ValidationException + + # These imports can't be placed in at top file level because it will cause a circular import in + # exceptions.py via _get_mfe_url_override + # Check for deprecated path input, either named or as first unnamed input + path = kwargs.pop("path", None) + if dest is None: + if path is not None: + dest = path + warnings.warn( + "the 'path' input for dump functions is deprecated. Please use 'dest' instead.", DeprecationWarning + ) + else: + msg = "No dump destination provided." + raise ValidationException( + message=msg, + no_personal_data_message="No dump destination Provided", + error_category=ErrorCategory.USER_ERROR, + target=ErrorTarget.GENERAL, + error_type=ValidationErrorType.MISSING_FIELD, + ) + + if isinstance(dest, (str, os.PathLike)): + try: + cm = open(dest, "w", encoding=DefaultOpenEncoding.WRITE) + except OSError as e: # FileNotFoundError introduced in Python 3 + msg = "No such parent folder path or not a file path: {}" + raise ValidationException( + message=msg.format(dest), + no_personal_data_message=msg.format("[file_path]"), + error_category=ErrorCategory.USER_ERROR, + target=ErrorTarget.GENERAL, + error_type=ValidationErrorType.FILE_OR_FOLDER_NOT_FOUND, + ) from e + else: + # dest is a subclass of IO + if not dest.writable(): # dest is misformatted stream or file + msg = "File Permissions Error: The already-open \n\n inputted file is not writable." + raise ValidationException( + message=msg, + no_personal_data_message="File Permissions error", + error_category=ErrorCategory.USER_ERROR, + target=ErrorTarget.GENERAL, + error_type=ValidationErrorType.CANNOT_PARSE, + ) + cm = nullcontext(enter_result=dest) + + with cm as f: + try: + dump_yaml(data_dict, f, default_flow_style=default_flow_style) + except yaml.YAMLError as e: + msg = f"Error while parsing yaml file \n\n {str(e)}" + raise ValidationException( + message=msg, + no_personal_data_message="Error while parsing yaml file", + error_category=ErrorCategory.USER_ERROR, + target=ErrorTarget.GENERAL, + error_type=ValidationErrorType.CANNOT_PARSE, + ) from e + + +def dict_eq(dict1: Dict[str, Any], dict2: Dict[str, Any]) -> bool: + """Compare two dictionaries. + + :param dict1: The first dictionary + :type dict1: Dict[str, Any] + :param dict2: The second dictionary + :type dict2: Dict[str, Any] + :return: True if the two dictionaries are equal, False otherwise + :rtype: bool + """ + if not dict1 and not dict2: + return True + return dict1 == dict2 + + +def xor(a: Any, b: Any) -> bool: + """XOR two values. + + :param a: The first value + :type a: Any + :param b: The second value + :type b: Any + :return: False if the two values are both True or both False, True otherwise + :rtype: bool + """ + return bool(a) != bool(b) + + +def is_url(value: Union[PathLike, str]) -> bool: + """Check if a string is a valid URL. + + :param value: The string to check + :type value: Union[PathLike, str] + :return: True if the string is a valid URL, False otherwise + :rtype: bool + """ + try: + result = urlparse(str(value)) + return all([result.scheme, result.netloc]) + except ValueError: + return False + + +# Resolve an URL to long form if it is an azureml short from datastore URL, otherwise return the same value +def resolve_short_datastore_url(value: Union[PathLike, str], workspace: OperationScope) -> str: + """Resolve an URL to long form if it is an azureml short from datastore URL, otherwise return the same value. + + :param value: The URL to resolve + :type value: Union[PathLike, str] + :param workspace: The workspace + :type workspace: OperationScope + :return: The resolved URL + :rtype: str + """ + from azure.ai.ml.exceptions import ValidationException + + # These imports can't be placed in at top file level because it will cause a circular import in + # exceptions.py via _get_mfe_url_override + + try: + # Check if the URL is an azureml URL + if urlparse(str(value)).scheme == "azureml": + from azure.ai.ml._utils._storage_utils import AzureMLDatastorePathUri + + data_store_path_uri = AzureMLDatastorePathUri(value) + if data_store_path_uri.uri_type == "Datastore": + return AzureMLDatastorePathUri(value).to_long_uri( + subscription_id=workspace.subscription_id, + resource_group_name=workspace.resource_group_name, + workspace_name=workspace.workspace_name, + ) + + except (ValueError, ValidationException): + pass + + # If the URL is not an valid URL (e.g. a local path) or not an azureml URL + # (e.g. a http URL), just return the same value + return value + + +def is_mlflow_uri(value: Union[PathLike, str]) -> bool: + """Check if a string is a valid mlflow uri. + + :param value: The string to check + :type value: Union[PathLike, str] + :return: True if the string is a valid mlflow uri, False otherwise + :rtype: bool + """ + try: + return urlparse(str(value)).scheme == "runs" + except ValueError: + return False + + +def validate_ml_flow_folder(path: str, model_type: string) -> None: + """Validate that the path is a valid ml flow folder. + + :param path: The path to validate + :type path: str + :param model_type: The model type + :type model_type: str + :return: No return value + :rtype: None + :raises ~azure.ai.ml.exceptions.ValidationException: Raised if the path is not a valid ml flow folder. + """ + from azure.ai.ml.exceptions import ErrorTarget, ValidationErrorType, ValidationException + + # These imports can't be placed in at top file level because it will cause a circular import in + # exceptions.py via _get_mfe_url_override + + if not isinstance(path, str): + path = path.as_posix() + path_array = path.split("/") + if model_type != "mlflow_model" or "." not in path_array[-1]: + return + msg = "Error with path {}. Model of type mlflow_model cannot register a file." + raise ValidationException( + message=msg.format(path), + no_personal_data_message=msg.format("[path]"), + target=ErrorTarget.MODEL, + error_type=ValidationErrorType.INVALID_VALUE, + ) + + +# modified from: https://stackoverflow.com/a/33245493/8093897 +def is_valid_uuid(test_uuid: str) -> bool: + """Check if a string is a valid UUID. + + :param test_uuid: The string to check + :type test_uuid: str + :return: True if the string is a valid UUID, False otherwise + :rtype: bool + """ + try: + uuid_obj = UUID(test_uuid, version=4) + except ValueError: + return False + return str(uuid_obj) == test_uuid + + +@singledispatch +def from_iso_duration_format(duration: Optional[Any] = None) -> int: # pylint: disable=unused-argument + """Convert ISO duration format to seconds. + + :param duration: The duration to convert + :type duration: Optional[Any] + :return: The converted duration + :rtype: int + """ + return None + + +@from_iso_duration_format.register(str) +def _(duration: str) -> int: + return int(isodate.parse_duration(duration).total_seconds()) + + +@from_iso_duration_format.register(timedelta) +def _(duration: timedelta) -> int: + return int(duration.total_seconds()) + + +def to_iso_duration_format_mins(time_in_mins: Optional[Union[int, float]]) -> str: + """Convert minutes to ISO duration format. + + :param time_in_mins: The time in minutes to convert + :type time_in_mins: Optional[Union[int, float]] + :return: The converted time in ISO duration format + :rtype: str + """ + return isodate.duration_isoformat(timedelta(minutes=time_in_mins)) if time_in_mins else None + + +def from_iso_duration_format_mins(duration: Optional[str]) -> int: + """Convert ISO duration format to minutes. + + :param duration: The duration to convert + :type duration: Optional[str] + :return: The converted duration + :rtype: int + """ + return int(from_iso_duration_format(duration) / 60) if duration else None + + +def to_iso_duration_format(time_in_seconds: Optional[Union[int, float]]) -> str: + """Convert seconds to ISO duration format. + + :param time_in_seconds: The time in seconds to convert + :type time_in_seconds: Optional[Union[int, float]] + :return: The converted time in ISO duration format + :rtype: str + """ + return isodate.duration_isoformat(timedelta(seconds=time_in_seconds)) if time_in_seconds else None + + +def to_iso_duration_format_ms(time_in_ms: Optional[Union[int, float]]) -> str: + """Convert milliseconds to ISO duration format. + + :param time_in_ms: The time in milliseconds to convert + :type time_in_ms: Optional[Union[int, float]] + :return: The converted time in ISO duration format + :rtype: str + """ + return isodate.duration_isoformat(timedelta(milliseconds=time_in_ms)) if time_in_ms else None + + +def from_iso_duration_format_ms(duration: Optional[str]) -> int: + """Convert ISO duration format to milliseconds. + + :param duration: The duration to convert + :type duration: Optional[str] + :return: The converted duration + :rtype: int + """ + return from_iso_duration_format(duration) * 1000 if duration else None + + +def to_iso_duration_format_days(time_in_days: Optional[int]) -> str: + """Convert days to ISO duration format. + + :param time_in_days: The time in days to convert + :type time_in_days: Optional[int] + :return: The converted time in ISO duration format + :rtype: str + """ + return isodate.duration_isoformat(timedelta(days=time_in_days)) if time_in_days else None + + +@singledispatch +def from_iso_duration_format_days(duration: Optional[Any] = None) -> int: # pylint: disable=unused-argument + return None + + +@from_iso_duration_format_days.register(str) +def _(duration: str) -> int: + return int(isodate.parse_duration(duration).days) + + +@from_iso_duration_format_days.register(timedelta) +def _(duration: timedelta) -> int: + return int(duration.days) + + +def _get_base_urls_from_discovery_service( + workspace_operations: "WorkspaceOperations", workspace_name: str, requests_pipeline: HttpPipeline +) -> Dict[WorkspaceDiscoveryUrlKey, str]: + """Fetch base urls for a workspace from the discovery service. + + :param WorkspaceOperations workspace_operations: + :param str workspace_name: The name of the workspace + :param HttpPipeline requests_pipeline: An HTTP pipeline to make requests with + :returns: A dictionary mapping url types to base urls + :rtype: Dict[WorkspaceDiscoveryUrlKey,str] + """ + discovery_url = workspace_operations.get(workspace_name).discovery_url + + return json.loads( + download_text_from_url( + discovery_url, + create_requests_pipeline_with_retry(requests_pipeline=requests_pipeline), + ) + ) + + +def _get_mfe_base_url_from_discovery_service( + workspace_operations: Any, workspace_name: str, requests_pipeline: HttpPipeline +) -> str: + all_urls = _get_base_urls_from_discovery_service(workspace_operations, workspace_name, requests_pipeline) + return f"{all_urls[WorkspaceDiscoveryUrlKey.API]}{MFE_PATH_PREFIX}" + + +def _get_mfe_base_url_from_registry_discovery_service( + workspace_operations: Any, workspace_name: str, requests_pipeline: HttpPipeline +) -> str: + all_urls = _get_base_urls_from_discovery_service(workspace_operations, workspace_name, requests_pipeline) + return all_urls[WorkspaceDiscoveryUrlKey.API] + + +def _get_workspace_base_url(workspace_operations: Any, workspace_name: str, requests_pipeline: HttpPipeline) -> str: + all_urls = _get_base_urls_from_discovery_service(workspace_operations, workspace_name, requests_pipeline) + return all_urls[WorkspaceDiscoveryUrlKey.API] + + +def _get_mfe_base_url_from_batch_endpoint(endpoint: "BatchEndpoint") -> str: + return endpoint.scoring_uri.split("/subscriptions/")[0] + + +# Allows to use a modified client with a provided url +@contextmanager +def modified_operation_client(operation_to_modify, url_to_use): + """Modify the operation client to use a different url. + + :param operation_to_modify: The operation to modify + :type operation_to_modify: Any + :param url_to_use: The url to use + :type url_to_use: str + :return: The modified operation + :rtype: Any + """ + original_api_base_url = None + try: + # Modify the operation + if url_to_use: + original_api_base_url = operation_to_modify._client._base_url + operation_to_modify._client._base_url = url_to_use + yield + finally: + # Undo the modification + if original_api_base_url: + operation_to_modify._client._base_url = original_api_base_url + + +def from_iso_duration_format_min_sec(duration: Optional[str]) -> str: + """Convert ISO duration format to min:sec format. + + :param duration: The duration to convert + :type duration: Optional[str] + :return: The converted duration + :rtype: str + """ + return duration.split(".")[0].replace("PT", "").replace("M", "m ") + "s" + + +def hash_dict(items: Dict[str, Any], keys_to_omit: Optional[Iterable[str]] = None) -> str: + """Return hash GUID of a dictionary except keys_to_omit. + + :param items: The dict to hash + :type items: Dict[str, Any] + :param keys_to_omit: Keys to omit before hashing + :type keys_to_omit: Optional[Iterable[str]] + :return: The hash GUID of the dictionary + :rtype: str + """ + if keys_to_omit is None: + keys_to_omit = [] + items = pydash.omit(items, keys_to_omit) + # serialize dict with order so same dict will have same content + serialized_component_interface = json.dumps(items, sort_keys=True) + object_hash = hashlib.md5() # nosec + object_hash.update(serialized_component_interface.encode("utf-8")) + return str(UUID(object_hash.hexdigest())) + + +def convert_identity_dict( + identity: Optional[ManagedServiceIdentity] = None, +) -> ManagedServiceIdentity: + """Convert identity to the right format. + + :param identity: The identity to convert + :type identity: Optional[ManagedServiceIdentity] + :return: The converted identity + :rtype: ManagedServiceIdentity + """ + if identity: + if identity.type.lower() in ("system_assigned", "none"): + identity = ManagedServiceIdentity(type="SystemAssigned") + else: + if identity.user_assigned_identities: + if isinstance(identity.user_assigned_identities, dict): # if the identity is already in right format + return identity + ids = {} + for id in identity.user_assigned_identities: # pylint: disable=redefined-builtin + ids[id["resource_id"]] = {} + identity.user_assigned_identities = ids + identity.type = snake_to_camel(identity.type) + else: + identity = ManagedServiceIdentity(type="SystemAssigned") + return identity + + +def strip_double_curly(io_binding_val: str) -> str: + """Strip double curly brackets from a string. + + :param io_binding_val: The string to strip + :type io_binding_val: str + :return: The string with double curly brackets stripped + :rtype: str + """ + return io_binding_val.replace("${{", "").replace("}}", "") + + +def append_double_curly(io_binding_val: str) -> str: + """Append double curly brackets to a string. + + :param io_binding_val: The string to append to + :type io_binding_val: str + :return: The string with double curly brackets appended + :rtype: str + """ + return f"${{{{{io_binding_val}}}}}" + + +def map_single_brackets_and_warn(command: str) -> str: + """Map single brackets to double brackets and warn if found. + + :param command: The command to map + :type command: str + :return: The mapped command + :rtype: str + """ + + def _check_for_parameter(param_prefix: str, command_string: str) -> Tuple[bool, str]: + template_prefix = r"(?<!\{)\{" + template_suffix = r"\.([^}]*)\}(?!\})" + template = template_prefix + param_prefix + template_suffix + should_warn = False + if bool(re.search(template, command_string)): + should_warn = True + command_string = re.sub(template, r"${{" + param_prefix + r".\g<1>}}", command_string) + return (should_warn, command_string) + + input_warn, command = _check_for_parameter("inputs", command) + output_warn, command = _check_for_parameter("outputs", command) + sweep_warn, command = _check_for_parameter("search_space", command) + if input_warn or output_warn or sweep_warn: + module_logger.warning("Use of {} for parameters is deprecated, instead use ${{}}.") + return command + + +def transform_dict_keys(data: Dict[str, Any], casing_transform: Callable[[str], str]) -> Dict[str, Any]: + """Convert all keys of a nested dictionary according to the passed casing_transform function. + + :param data: The data to transform + :type data: Dict[str, Any] + :param casing_transform: A callable applied to all keys in data + :type casing_transform: Callable[[str], str] + :return: A dictionary with transformed keys + :rtype: dict + """ + return { + casing_transform(key): transform_dict_keys(val, casing_transform) if isinstance(val, dict) else val + for key, val in data.items() + } + + +def merge_dict(origin, delta, dep=0) -> dict: + """Merge two dicts recursively. + Note that the function will return a copy of the origin dict if the depth of the recursion is 0. + + :param origin: The original dictionary + :type origin: dict + :param delta: The delta dictionary + :type delta: dict + :param dep: The depth of the recursion + :type dep: int + :return: The merged dictionary + :rtype: dict + """ + result = copy.deepcopy(origin) if dep == 0 else origin + for key, val in delta.items(): + origin_val = origin.get(key) + # Merge delta dict with original dict + if isinstance(origin_val, dict) and isinstance(val, dict): + result[key] = merge_dict(origin_val, val, dep + 1) + continue + result[key] = copy.deepcopy(val) + return result + + +def retry( + exceptions: Union[Tuple[Exception], Exception], + failure_msg: str, + logger: Any, + max_attempts: int = 1, + delay_multiplier: int = 0.25, +) -> Callable: + """Retry a function if it fails. + + :param exceptions: Exceptions to retry on. + :type exceptions: Union[Tuple[Exception], Exception] + :param failure_msg: Message to log on failure. + :type failure_msg: str + :param logger: Logger to use. + :type logger: Any + :param max_attempts: Maximum number of attempts. + :type max_attempts: int + :param delay_multiplier: Multiplier for delay between attempts. + :type delay_multiplier: int + :return: Decorated function. + :rtype: Callable + """ + + def retry_decorator(f): + @wraps(f) + def func_with_retries(*args, **kwargs): # pylint: disable=inconsistent-return-statements + tries = max_attempts + 1 + counter = 1 + while tries > 1: + delay = delay_multiplier * 2**counter + random.uniform(0, 1) + try: + return f(*args, **kwargs) + except exceptions as e: + tries -= 1 + counter += 1 + if tries == 1: + logger.warning(failure_msg) + raise e + logger.info(f"Operation failed. Retrying in {delay} seconds.") + time.sleep(delay) + + return func_with_retries + + return retry_decorator + + +def get_list_view_type(include_archived: bool, archived_only: bool) -> ListViewType: + """Get the list view type based on the include_archived and archived_only flags. + + :param include_archived: Whether to include archived items. + :type include_archived: bool + :param archived_only: Whether to only include archived items. + :type archived_only: bool + :return: The list view type. + :rtype: ListViewType + """ + if include_archived and archived_only: + msg = "Cannot provide both archived-only and include-archived." + raise MlException(message=msg, no_personal_data_message=msg) + if include_archived: + return ListViewType.ALL + if archived_only: + return ListViewType.ARCHIVED_ONLY + return ListViewType.ACTIVE_ONLY + + +def is_data_binding_expression( + value: str, binding_prefix: Union[str, List[str]] = "", is_singular: bool = True +) -> bool: + """Check if a value is a data-binding expression with specific binding target(prefix). Note that the function will + return False if the value is not a str. For example, if binding_prefix is ["parent", "jobs"], then input_value is a + data-binding expression only if the binding target starts with "parent.jobs", like "${{parent.jobs.xxx}}" if + is_singular is False, return True even if input_value includes non-binding part or multiple binding targets, like + "${{parent.jobs.xxx}}_extra" and "${{parent.jobs.xxx}}_{{parent.jobs.xxx}}". + + :param value: Value to check. + :type value: str + :param binding_prefix: Prefix to check for. + :type binding_prefix: Union[str, List[str]] + :param is_singular: should the value be a singular data-binding expression, like "${{parent.jobs.xxx}}". + :type is_singular: bool + :return: True if the value is a data-binding expression, False otherwise. + :rtype: bool + """ + return len(get_all_data_binding_expressions(value, binding_prefix, is_singular)) > 0 + + +def get_all_data_binding_expressions( + value: str, binding_prefix: Union[str, List[str]] = "", is_singular: bool = True +) -> List[str]: + """Get all data-binding expressions in a value with specific binding target(prefix). Note that the function will + return an empty list if the value is not a str. + + :param value: Value to extract. + :type value: str + :param binding_prefix: Prefix to filter. + :type binding_prefix: Union[str, List[str]] + :param is_singular: should the value be a singular data-binding expression, like "${{parent.jobs.xxx}}". + :type is_singular: bool + :return: list of data-binding expressions. + :rtype: List[str] + """ + if isinstance(binding_prefix, str): + binding_prefix = [binding_prefix] + if isinstance(value, str): + target_regex = r"\$\{\{\s*(" + "\\.".join(binding_prefix) + r"\S*?)\s*\}\}" + if is_singular: + target_regex = "^" + target_regex + "$" + return re.findall(target_regex, value) + return [] + + +def is_private_preview_enabled(): + """Check if private preview features are enabled. + + :return: True if private preview features are enabled, False otherwise. + :rtype: bool + """ + return os.getenv(AZUREML_PRIVATE_FEATURES_ENV_VAR) in ["True", "true", True] + + +def is_bytecode_optimization_enabled(): + """Check if bytecode optimization is enabled: + 1) bytecode package is installed + 2) private preview is enabled + 3) python version is between 3.6 and 3.11 + + :return: True if bytecode optimization is enabled, False otherwise. + :rtype: bool + """ + try: + import bytecode # pylint: disable=unused-import + + return is_private_preview_enabled() and (3, 6) < sys.version_info < (3, 12) + except ImportError: + return False + + +def is_on_disk_cache_enabled(): + """Check if on-disk cache for component registrations in pipeline submission is enabled. + + :return: True if on-disk cache is enabled, False otherwise. + :rtype: bool + """ + return os.getenv(AZUREML_DISABLE_ON_DISK_CACHE_ENV_VAR) not in ["True", "true", True] + + +def is_concurrent_component_registration_enabled(): # pylint: disable=name-too-long + """Check if concurrent component registrations in pipeline submission is enabled. + + :return: True if concurrent component registration is enabled, False otherwise. + :rtype: bool + """ + return os.getenv(AZUREML_DISABLE_CONCURRENT_COMPONENT_REGISTRATION) not in ["True", "true", True] + + +def _is_internal_components_enabled(): + return os.getenv(AZUREML_INTERNAL_COMPONENTS_ENV_VAR) in ["True", "true", True] + + +def try_enable_internal_components(*, force=False) -> bool: + """Try to enable internal components for the current process. This is the only function outside _internal that + references _internal. + + :keyword force: Force enable internal components even if enabled before. + :type force: bool + :return: True if internal components are enabled, False otherwise. + :rtype: bool + """ + if _is_internal_components_enabled(): + from azure.ai.ml._internal import enable_internal_components_in_pipeline + + enable_internal_components_in_pipeline(force=force) + + return True + return False + + +def is_internal_component_data(data: Dict[str, Any], *, raise_if_not_enabled: bool = False) -> bool: + """Check if the data is an internal component data by checking schema url prefix. + + :param data: The data to check. + :type data: Dict[str, Any] + :keyword raise_if_not_enabled: Raise exception if the data is an internal component data but + internal components is not enabled. + :type raise_if_not_enabled: bool + :return: True if the data is an internal component data, False otherwise. + :rtype: bool + :raises ~azure.ai.ml.exceptions.ValidationException: Raised if the data is an internal component data but + internal components is not enabled. + """ + from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationErrorType, ValidationException + + # These imports can't be placed in at top file level because it will cause a circular import in + # exceptions.py via _get_mfe_url_override + + schema = data.get(CommonYamlFields.SCHEMA, None) + + if schema is None or not isinstance(schema, str): + return False + + if not schema.startswith(AZUREML_INTERNAL_COMPONENTS_SCHEMA_PREFIX): + return False + + if not _is_internal_components_enabled() and raise_if_not_enabled: + no_personal_data_message = ( + f"Internal components is a private feature in v2, please set environment variable " + f"{AZUREML_INTERNAL_COMPONENTS_ENV_VAR} to true to use it." + ) + msg = f"Detected schema url {schema}. {no_personal_data_message}" + raise ValidationException( + message=msg, + target=ErrorTarget.COMPONENT, + error_type=ValidationErrorType.INVALID_VALUE, + no_personal_data_message=no_personal_data_message, + error_category=ErrorCategory.USER_ERROR, + ) + + return True + + +def is_valid_node_name(name: str) -> bool: + """Check if `name` is a valid node name + + :param name: A node name + :type name: str + :return: Return True if the string is a valid Python identifier in lower ASCII range, False otherwise. + :rtype: bool + """ + return isinstance(name, str) and name.isidentifier() and re.fullmatch(r"^[a-z_][a-z\d_]*", name) is not None + + +def parse_args_description_from_docstring(docstring: str) -> Dict[str, str]: + """Return arg descriptions in docstring with google style. + + e.g. + docstring = + ''' + A pipeline with detailed docstring, including descriptions for inputs and outputs. + + In this pipeline docstring, there are descriptions for inputs and outputs + Input/Output descriptions can infer from descriptions here. + + Args: + job_in_path: a path parameter + job_in_number: a number parameter + with multi-line description + job_in_int (int): a int parameter + + Other docstring xxxxxx + random_key: random_value + ''' + + return dict: + args = { + 'job_in_path': 'a path parameter', + 'job_in_number': 'a number parameter with multi-line description', + 'job_in_int': 'a int parameter' + } + + :param docstring: A Google-style docstring + :type docstring: str + :return: A map of parameter names to parameter descriptions + :rtype: Dict[str, str] + """ + args = {} + if not isinstance(docstring, str): + return args + lines = [line.strip() for line in docstring.splitlines()] + for index, line in enumerate(lines): + if line.lower() == "args:": + args_region = lines[index + 1 :] + args_line_end = args_region.index("") if "" in args_region else len(args_region) + args_region = args_region[0:args_line_end] + while len(args_region) > 0 and ":" in args_region[0]: + arg_line = args_region[0] + colon_index = arg_line.index(":") + arg, description = ( + arg_line[0:colon_index].strip(), + arg_line[colon_index + 1 :].strip(), + ) + # handle case like "param (float) : xxx" + if "(" in arg: + arg = arg[0 : arg.index("(")].strip() + args[arg] = description + args_region.pop(0) + # handle multi-line description, assuming description has no colon inside. + while len(args_region) > 0 and ":" not in args_region[0]: + args[arg] += " " + args_region[0] + args_region.pop(0) + return args + + +def convert_windows_path_to_unix(path: Union[str, PathLike]) -> str: + """Convert a Windows path to a Unix path. + + :param path: A Windows path + :type path: Union[str, os.PathLike] + :return: A Unix path + :rtype: str + """ + return PureWindowsPath(path).as_posix() + + +def _is_user_error_from_status_code(http_status_code): + return 400 <= http_status_code < 500 + + +def _str_to_bool(s: str) -> bool: + """Converts a string to a boolean + + Can be used as a type for argument in argparse, return argument's boolean value according to it's literal value. + + :param s: The string to convert + :type s: str + :return: True if s is "true" (case-insensitive), otherwise returns False. + :rtype: bool + """ + if not isinstance(s, str): + return False + return s.lower() == "true" + + +def _is_user_error_from_exception_type(e: Optional[Exception]) -> bool: + """Determine whether if an exception is user error from it's exception type. + + :param e: An exception + :type e: Optional[Exception] + :return: True if exception is a user error + :rtype: bool + """ + # Connection error happens on user's network failure, should be user error. + # For OSError/IOError with error no 28: "No space left on device" should be sdk user error + return isinstance(e, (ConnectionError, KeyboardInterrupt)) or (isinstance(e, (IOError, OSError)) and e.errno == 28) + + +class DockerProxy: + """A proxy class for docker module. It will raise a more user-friendly error message if docker module is not + installed. + """ + + def __getattribute__(self, name: str) -> Any: + try: + import docker + + return getattr(docker, name) + except ModuleNotFoundError as e: + msg = "Please install docker in the current python environment with `pip install docker` and try again." + raise MlException(message=msg, no_personal_data_message=msg) from e + + +def get_all_enum_values_iter(enum_type: type) -> Iterable[Any]: + """Get all values of an enum type. + + :param enum_type: An "enum" (not necessary enum.Enum) + :type enum_type: Type + :return: An iterable of all of the attributes of `enum_type` + :rtype: Iterable[Any] + """ + for key in dir(enum_type): + if not key.startswith("_"): + yield getattr(enum_type, key) + + +def write_to_shared_file(file_path: Union[str, PathLike], content: str): + """Open file with specific mode and return the file object. + + :param file_path: Path to the file. + :type file_path: Union[str, os.PathLike] + :param content: Content to write to the file. + :type content: str + """ + with open(file_path, "w", encoding=DefaultOpenEncoding.WRITE) as f: + f.write(content) + + # share_mode means read/write for owner, group and others + share_mode, mode_mask = 0o666, 0o777 + if os.stat(file_path).st_mode & mode_mask != share_mode: + try: + os.chmod(file_path, share_mode) + except PermissionError: + pass + + +def _get_valid_dot_keys_with_wildcard_impl( + left_reversed_parts, root, *, validate_func=None, cur_node=None, processed_parts=None +) -> List[str]: + if len(left_reversed_parts) == 0: + if validate_func is None or validate_func(root, processed_parts): + return [".".join(processed_parts)] + return [] + + if cur_node is None: + cur_node = root + if not isinstance(cur_node, dict): + return [] + if processed_parts is None: + processed_parts = [] + + key: str = left_reversed_parts.pop() + result = [] + if key == "*": + for next_key in cur_node: + if not isinstance(next_key, str): + continue + processed_parts.append(next_key) + result.extend( + _get_valid_dot_keys_with_wildcard_impl( + left_reversed_parts, + root, + validate_func=validate_func, + cur_node=cur_node[next_key], + processed_parts=processed_parts, + ) + ) + processed_parts.pop() + elif key in cur_node: + processed_parts.append(key) + result = _get_valid_dot_keys_with_wildcard_impl( + left_reversed_parts, + root, + validate_func=validate_func, + cur_node=cur_node[key], + processed_parts=processed_parts, + ) + processed_parts.pop() + + left_reversed_parts.append(key) + return result + + +def get_valid_dot_keys_with_wildcard( + root: Dict[str, Any], + dot_key_wildcard: str, + *, + validate_func: Optional[Callable[[List[str], Dict[str, Any]], bool]] = None, +) -> List[str]: + """Get all valid dot keys with wildcard. Only "x.*.x" and "x.*" is supported for now. + + A valid dot key should satisfy the following conditions: + 1) It should be a valid dot key in the root node. + 2) It should satisfy the validation function. + + :param root: Root node. + :type root: Dict[str, Any] + :param dot_key_wildcard: Dot key with wildcard, e.g. "a.*.c". + :type dot_key_wildcard: str + :keyword validate_func: Validation function. It takes two parameters: the root node and the dot key parts. + If None, no validation will be performed. + :paramtype validate_func: Optional[Callable[[List[str], Dict[str, Any]], bool]] + :return: List of valid dot keys. + :rtype: List[str] + """ + left_reversed_parts = dot_key_wildcard.split(".")[::-1] + return _get_valid_dot_keys_with_wildcard_impl(left_reversed_parts, root, validate_func=validate_func) + + +def get_base_directory_for_cache() -> Path: + """Get the base directory for cache files. + + :return: The base directory for cache files. + :rtype: Path + """ + return Path(tempfile.gettempdir()).joinpath("azure-ai-ml") + + +def get_versioned_base_directory_for_cache() -> Path: + """Get the base directory for cache files of current version of azure-ai-ml. + Cache files of different versions will be stored in different directories. + + :return: The base directory for cache files of current version of azure-ai-ml. + :rtype: Path + """ + # import here to avoid circular import + from azure.ai.ml._version import VERSION + + return get_base_directory_for_cache().joinpath(VERSION) + + +# pylint: disable-next=name-too-long +def get_resource_and_group_name_from_resource_id(armstr: str) -> str: + if armstr.find("/") == -1: + return armstr, None + return armstr.split("/")[-1], armstr.split("/")[-5] + + +# pylint: disable-next=name-too-long +def get_resource_group_name_from_resource_group_id(armstr: str) -> str: + if armstr.find("/") == -1: + return armstr + return armstr.split("/")[-1] + + +def extract_name_and_version(azureml_id: str) -> Dict[str, str]: + """Extract name and version from azureml id. + + :param azureml_id: AzureML id. + :type azureml_id: str + :return: A dict of name and version. + :rtype: Dict[str, str] + """ + if not isinstance(azureml_id, str): + raise ValueError("azureml_id should be a string but got {}: {}.".format(type(azureml_id), azureml_id)) + if azureml_id.count(":") != 1: + raise ValueError("azureml_id should be in the format of name:version but got {}.".format(azureml_id)) + name, version = azureml_id.split(":") + return { + "name": name, + "version": version, + } + + +def _get_evaluator_properties(): + return {"is-promptflow": "true", "is-evaluator": "true"} + + +def _is_evaluator(properties: Dict[str, str]) -> bool: + return properties.get("is-evaluator") == "true" and properties.get("is-promptflow") == "true" |