path: root/.venv/lib/python3.12/site-packages/azure/ai/ml/entities/_deployment/online_deployment.py

# ---------------------------------------------------------
# Copyright (c) Microsoft Corporation. All rights reserved.
# ---------------------------------------------------------

# pylint: disable=protected-access,arguments-renamed,unidiomatic-typecheck

import logging
import os
import typing
from abc import abstractmethod
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple, Union, cast

from azure.ai.ml._restclient.v2023_04_01_preview.models import CodeConfiguration as RestCodeConfiguration
from azure.ai.ml._restclient.v2023_04_01_preview.models import EndpointComputeType
from azure.ai.ml._restclient.v2023_04_01_preview.models import (
    KubernetesOnlineDeployment as RestKubernetesOnlineDeployment,
)
from azure.ai.ml._restclient.v2023_04_01_preview.models import ManagedOnlineDeployment as RestManagedOnlineDeployment
from azure.ai.ml._restclient.v2023_04_01_preview.models import OnlineDeployment as RestOnlineDeploymentData
from azure.ai.ml._restclient.v2023_04_01_preview.models import OnlineDeploymentProperties as RestOnlineDeploymentDetails
from azure.ai.ml._restclient.v2023_04_01_preview.models import Sku as RestSku
from azure.ai.ml._schema._deployment.online.online_deployment import (
    KubernetesOnlineDeploymentSchema,
    ManagedOnlineDeploymentSchema,
)
from azure.ai.ml._utils._arm_id_utils import _parse_endpoint_name_from_deployment_id
from azure.ai.ml._utils.utils import camel_to_snake
from azure.ai.ml.constants._common import BASE_PATH_CONTEXT_KEY, PARAMS_OVERRIDE_KEY, TYPE, ArmConstants
from azure.ai.ml.constants._endpoint import EndpointYamlFields
from azure.ai.ml.entities._assets import Code
from azure.ai.ml.entities._assets._artifacts.model import Model
from azure.ai.ml.entities._assets.environment import Environment
from azure.ai.ml.entities._deployment.code_configuration import CodeConfiguration
from azure.ai.ml.entities._deployment.data_collector import DataCollector
from azure.ai.ml.entities._deployment.deployment_settings import OnlineRequestSettings, ProbeSettings
from azure.ai.ml.entities._deployment.resource_requirements_settings import ResourceRequirementsSettings
from azure.ai.ml.entities._deployment.scale_settings import (
    DefaultScaleSettings,
    OnlineScaleSettings,
    TargetUtilizationScaleSettings,
)
from azure.ai.ml.entities._endpoint._endpoint_helpers import validate_endpoint_or_deployment_name
from azure.ai.ml.entities._util import load_from_dict
from azure.ai.ml.exceptions import (
    DeploymentException,
    ErrorCategory,
    ErrorTarget,
    ValidationErrorType,
    ValidationException,
)

from .deployment import Deployment

module_logger = logging.getLogger(__name__)


# pylint: disable=too-many-instance-attributes
class OnlineDeployment(Deployment):
    """Online endpoint deployment entity.

    :param name: Name of the deployment resource.
    :type name: str
    :param endpoint_name: Name of the endpoint resource, defaults to None
    :type endpoint_name: typing.Optional[str]
    :param tags: Tag dictionary. Tags can be added, removed, and updated, defaults to None
    :type tags: typing.Optional[typing.Dict[str, typing.Any]]
    :param properties: The asset property dictionary, defaults to None
    :type properties: typing.Optional[typing.Dict[str, typing.Any]]
    :param description: Description of the resource, defaults to None
    :type description: typing.Optional[str]
    :param model: Model entity for the endpoint deployment, defaults to None
    :type model: typing.Optional[typing.Union[str, ~azure.ai.ml.entities.Model]]
    :param data_collector: Data Collector entity for the endpoint deployment, defaults to None
    :type data_collector: typing.Optional[typing.Union[str, ~azure.ai.ml.entities.DataCollector]]
    :param code_configuration: Code Configuration, defaults to None
    :type code_configuration: typing.Optional[~azure.ai.ml.entities.CodeConfiguration]
    :param environment: Environment entity for the endpoint deployment, defaults to None
    :type environment: typing.Optional[typing.Union[str, ~azure.ai.ml.entities.Environment]]
    :param app_insights_enabled: Is appinsights enabled, defaults to False
    :type app_insights_enabled: typing.Optional[bool]
    :param scale_settings: How the online deployment will scale, defaults to None
    :type scale_settings: typing.Optional[~azure.ai.ml.entities.OnlineScaleSettings]
    :param request_settings: Online Request Settings, defaults to None
    :type request_settings: typing.Optional[~azure.ai.ml.entities.OnlineRequestSettings]
    :param liveness_probe: Liveness probe settings, defaults to None
    :type liveness_probe: typing.Optional[~azure.ai.ml.entities.ProbeSettings]
    :param readiness_probe: Readiness probe settings, defaults to None
    :type readiness_probe: typing.Optional[~azure.ai.ml.entities.ProbeSettings]
    :param environment_variables: Environment variables that will be set in deployment, defaults to None
    :type environment_variables: typing.Optional[typing.Dict[str, str]]
    :param instance_count: The instance count used for this deployment, defaults to None
    :type instance_count: typing.Optional[int]
    :param instance_type: Azure compute sku, defaults to None
    :type instance_type: typing.Optional[str]
    :param model_mount_path: The path to mount the model in custom container, defaults to None
    :type model_mount_path: typing.Optional[str]
    :param code_path: Equivalent to code_configuration.code, will be ignored if code_configuration is present
        , defaults to None
    :type code_path: typing.Optional[typing.Union[str, os.PathLike]]
    :param scoring_script: Equivalent to code_configuration.code.scoring_script.
        Will be ignored if code_configuration is present, defaults to None
    :type scoring_script: typing.Optional[typing.Union[str, os.PathLike]]
    """

    def __init__(
        self,
        name: str,
        *,
        endpoint_name: Optional[str] = None,
        tags: Optional[Dict[str, typing.Any]] = None,
        properties: Optional[Dict[str, typing.Any]] = None,
        description: Optional[str] = None,
        model: Optional[Union[str, "Model"]] = None,
        data_collector: Optional[DataCollector] = None,
        code_configuration: Optional[CodeConfiguration] = None,
        environment: Optional[Union[str, "Environment"]] = None,
        app_insights_enabled: Optional[bool] = False,
        scale_settings: Optional[OnlineScaleSettings] = None,
        request_settings: Optional[OnlineRequestSettings] = None,
        liveness_probe: Optional[ProbeSettings] = None,
        readiness_probe: Optional[ProbeSettings] = None,
        environment_variables: Optional[Dict[str, str]] = None,
        instance_count: Optional[int] = None,
        instance_type: Optional[str] = None,
        model_mount_path: Optional[str] = None,
        code_path: Optional[Union[str, os.PathLike]] = None,  # promoted property from code_configuration.code
        scoring_script: Optional[Union[str, os.PathLike]] = None,  # promoted property code_configuration.scoring_script
        **kwargs: typing.Any,
    ):
        self._provisioning_state: Optional[str] = kwargs.pop("provisioning_state", None)

        super(OnlineDeployment, self).__init__(
            name=name,
            endpoint_name=endpoint_name,
            tags=tags,
            properties=properties,
            description=description,
            model=model,
            code_configuration=code_configuration,
            environment=environment,
            environment_variables=environment_variables,
            code_path=code_path,
            scoring_script=scoring_script,
            **kwargs,
        )

        self.app_insights_enabled = app_insights_enabled
        self.scale_settings = scale_settings
        self.request_settings = request_settings
        self.liveness_probe = liveness_probe
        self.readiness_probe = readiness_probe
        self.instance_count = instance_count
        self._arm_type = ArmConstants.ONLINE_DEPLOYMENT_TYPE
        self.model_mount_path = model_mount_path
        self.instance_type = instance_type
        self.data_collector: Any = data_collector

    @property
    def provisioning_state(self) -> Optional[str]:
        """Deployment provisioning state, readonly.

        :return: Deployment provisioning state.
        :rtype: typing.Optional[str]
        """
        return self._provisioning_state

    def _generate_dependencies(self) -> Tuple:
        """Convert dependencies into ARM id or REST wrapper.

        :return: A 3-tuple of the code configuration, environment ID, and model ID.
        :rtype: Tuple[RestCodeConfiguration, str, str]
        """
        code = None

        if self.code_configuration:
            self.code_configuration._validate()
            if self.code_configuration.code is not None:
                if isinstance(self.code_configuration.code, str):
                    code_id = self.code_configuration.code
                elif not isinstance(self.code_configuration.code, os.PathLike):
                    code_id = self.code_configuration.code.id

                code = RestCodeConfiguration(
                    code_id=code_id,  # pylint: disable=possibly-used-before-assignment
                    scoring_script=self.code_configuration.scoring_script,
                )

        model_id = None
        if self.model:
            model_id = self.model if isinstance(self.model, str) else self.model.id

        environment_id = None
        if self.environment:
            environment_id = self.environment if isinstance(self.environment, str) else self.environment.id

        return code, environment_id, model_id

    @abstractmethod
    def _to_dict(self) -> Dict:
        pass

    @abstractmethod
    def _to_arm_resource_param(self, **kwargs: Any) -> Dict:
        pass

    @abstractmethod
    def _to_rest_object(self) -> RestOnlineDeploymentData:
        pass

    @classmethod
    def _from_rest_object(cls, deployment: RestOnlineDeploymentData) -> RestOnlineDeploymentDetails:
        if deployment.properties.endpoint_compute_type == EndpointComputeType.KUBERNETES:
            return KubernetesOnlineDeployment._from_rest_object(deployment)
        if deployment.properties.endpoint_compute_type == EndpointComputeType.MANAGED:
            return ManagedOnlineDeployment._from_rest_object(deployment)

        msg = f"Unsupported online endpoint type {deployment.properties.endpoint_compute_type}."
        raise DeploymentException(
            message=msg,
            target=ErrorTarget.ONLINE_DEPLOYMENT,
            no_personal_data_message=msg,
            error_category=ErrorCategory.SYSTEM_ERROR,
        )

    def _get_arm_resource(self, **kwargs: Any) -> Dict:
        resource: dict = super(OnlineDeployment, self)._get_arm_resource(**kwargs)
        depends_on = []
        if self.environment and isinstance(self.environment, Environment):
            depends_on.append(f"{self.environment._arm_type}Deployment")
        if self.code_configuration and self.code_configuration.code and isinstance(self.code_configuration.code, Code):
            depends_on.append(f"{self.code_configuration.code._arm_type}Deployment")
        if self.model and isinstance(self.model, Model):
            depends_on.append(f"{self.model._arm_type}Deployment")
        resource[ArmConstants.DEPENDSON_PARAMETER_NAME] = depends_on
        return resource

    def _get_arm_resource_and_params(self, **kwargs: Any) -> List:
        resource_param_tuple_list = [(self._get_arm_resource(**kwargs), self._to_arm_resource_param(**kwargs))]
        if self.environment and isinstance(self.environment, Environment):
            resource_param_tuple_list.extend(self.environment._get_arm_resource_and_params())
        if self.code_configuration and self.code_configuration.code and isinstance(self.code_configuration.code, Code):
            resource_param_tuple_list.extend(self.code_configuration.code._get_arm_resource_and_params())
        if self.model and isinstance(self.model, Model):
            resource_param_tuple_list.extend(self.model._get_arm_resource_and_params())
        return resource_param_tuple_list

    def _validate_name(self) -> None:
        if self.name:
            validate_endpoint_or_deployment_name(self.name, is_deployment=True)

    def _merge_with(self, other: Any) -> None:
        if other:
            if self.name != other.name:
                msg = "The deployment name: {} and {} are not matched when merging."
                raise ValidationException(
                    message=msg.format(self.name, other.name),
                    target=ErrorTarget.ONLINE_DEPLOYMENT,
                    no_personal_data_message=msg.format("[name1]", "[name2]"),
                    error_category=ErrorCategory.USER_ERROR,
                    error_type=ValidationErrorType.INVALID_VALUE,
                )
            super()._merge_with(other)
            self.app_insights_enabled = other.app_insights_enabled or self.app_insights_enabled
            # Adding noqa: Fix E721 do not compare types, use 'isinstance()'
            # isinstance will include checking for subclasses, which is explicitly undesired by a logic.
            if self.scale_settings and type(self.scale_settings) == type(other.scale_settings):  # noqa
                self.scale_settings._merge_with(other.scale_settings)
            else:
                self.scale_settings = other.scale_settings
            if self.request_settings:
                self.request_settings._merge_with(other.request_settings)
            else:
                self.request_settings = other.request_settings
            if self.liveness_probe:
                self.liveness_probe._merge_with(other.liveness_probe)
            else:
                self.liveness_probe = other.liveness_probe
            if self.readiness_probe:
                self.readiness_probe._merge_with(other.readiness_probe)
            else:
                self.readiness_probe = other.readiness_probe
            self.instance_count = other.instance_count or self.instance_count
            self.instance_type = other.instance_type or self.instance_type

    @classmethod
    def _set_scale_settings(cls, data: dict) -> None:
        if not hasattr(data, EndpointYamlFields.SCALE_SETTINGS):
            return

        scale_settings = data[EndpointYamlFields.SCALE_SETTINGS]
        keyName = TYPE
        if scale_settings and scale_settings[keyName] == "default":
            scale_copy = scale_settings.copy()
            for key in scale_copy:
                if key != keyName:
                    scale_settings.pop(key, None)

    @classmethod
    def _load(
        cls,
        data: Optional[Dict] = None,
        yaml_path: Optional[Union[os.PathLike, str]] = None,
        params_override: Optional[list] = None,
        **kwargs: Any,
    ) -> "OnlineDeployment":
        data = data or {}
        params_override = params_override or []
        context = {
            BASE_PATH_CONTEXT_KEY: Path(yaml_path).parent if yaml_path else Path.cwd(),
            PARAMS_OVERRIDE_KEY: params_override,
        }

        deployment_type = data.get("type", None)

        if deployment_type == camel_to_snake(EndpointComputeType.KUBERNETES.value):
            res_kub: OnlineDeployment = load_from_dict(KubernetesOnlineDeploymentSchema, data, context, **kwargs)
            return res_kub

        res_manage: OnlineDeployment = load_from_dict(ManagedOnlineDeploymentSchema, data, context, **kwargs)
        return res_manage


class KubernetesOnlineDeployment(OnlineDeployment):
    """Kubernetes Online endpoint deployment entity.

    :param name: Name of the deployment resource.
    :type name: str
    :param endpoint_name: Name of the endpoint resource, defaults to None
    :type endpoint_name: typing.Optional[str]
    :param tags: Tag dictionary. Tags can be added, removed, and updated., defaults to None
    :type tags: typing.Optional[typing.Dict[str, typing.Any]]
    :param properties: The asset property dictionary, defaults to None
    :type properties: typing.Optional[typing.Dict[str, typing.Any]]
    :param description: Description of the resource, defaults to None
    :type description: typing.Optional[str]
    :param model: Model entity for the endpoint deployment, defaults to None
    :type model: typing.Optional[typing.Union[str, ~azure.ai.ml.entities.Model]]
    :param code_configuration: Code Configuration, defaults to None
    :type code_configuration: typing.Optional[~azure.ai.ml.entities.CodeConfiguration]
    :param environment: Environment entity for the endpoint deployment, defaults to None
    :type environment: typing.Optional[typing.Union[str, ~azure.ai.ml.entities.Environment]]
    :param app_insights_enabled: Is appinsights enabled, defaults to False
    :type app_insights_enabled: bool
    :param scale_settings: How the online deployment will scale, defaults to None
    :type scale_settings: typing.Optional[typing.Union[~azure.ai.ml.entities.DefaultScaleSettings
        , ~azure.ai.ml.entities.TargetUtilizationScaleSettings]]
    :param request_settings: Online Request Settings, defaults to None
    :type request_settings: typing.Optional[OnlineRequestSettings]
    :param liveness_probe: Liveness probe settings, defaults to None
    :type liveness_probe: typing.Optional[~azure.ai.ml.entities.ProbeSettings]
    :param readiness_probe: Readiness probe settings, defaults to None
    :type readiness_probe: typing.Optional[~azure.ai.ml.entities.ProbeSettings]
    :param environment_variables: Environment variables that will be set in deployment, defaults to None
    :type environment_variables: typing.Optional[typing.Dict[str, str]]
    :param resources: Resource requirements settings, defaults to None
    :type resources: typing.Optional[~azure.ai.ml.entities.ResourceRequirementsSettings]
    :param instance_count: The instance count used for this deployment, defaults to None
    :type instance_count: typing.Optional[int]
    :param instance_type: The instance type defined by K8S cluster admin, defaults to None
    :type instance_type: typing.Optional[str]
    :param code_path: Equivalent to code_configuration.code, will be ignored if code_configuration is present
        , defaults to None
    :type code_path: typing.Optional[typing.Union[str, os.PathLike]]
    :param scoring_script: Equivalent to code_configuration.code.scoring_script.
        Will be ignored if code_configuration is present, defaults to None
    :type scoring_script: typing.Optional[typing.Union[str, os.PathLike]]
    """

    def __init__(
        self,
        *,
        name: str,
        endpoint_name: Optional[str] = None,
        tags: Optional[Dict[str, typing.Any]] = None,
        properties: Optional[Dict[str, typing.Any]] = None,
        description: Optional[str] = None,
        model: Optional[Union[str, "Model"]] = None,
        code_configuration: Optional[CodeConfiguration] = None,
        environment: Optional[Union[str, "Environment"]] = None,
        app_insights_enabled: bool = False,
        scale_settings: Optional[Union[DefaultScaleSettings, TargetUtilizationScaleSettings]] = None,
        request_settings: Optional[OnlineRequestSettings] = None,
        liveness_probe: Optional[ProbeSettings] = None,
        readiness_probe: Optional[ProbeSettings] = None,
        environment_variables: Optional[Dict[str, str]] = None,
        resources: Optional[ResourceRequirementsSettings] = None,
        instance_count: Optional[int] = None,
        instance_type: Optional[str] = None,
        code_path: Optional[Union[str, os.PathLike]] = None,  # promoted property from code_configuration.code
        scoring_script: Optional[
            Union[str, os.PathLike]
        ] = None,  # promoted property from code_configuration.scoring_script
        **kwargs: Any,
    ):
        kwargs["type"] = EndpointComputeType.KUBERNETES.value
        super(KubernetesOnlineDeployment, self).__init__(
            name=name,
            endpoint_name=endpoint_name,
            tags=tags,
            properties=properties,
            description=description,
            model=model,
            code_configuration=code_configuration,
            environment=environment,
            environment_variables=environment_variables,
            instance_count=instance_count,
            instance_type=instance_type,
            app_insights_enabled=app_insights_enabled,
            scale_settings=scale_settings,
            request_settings=request_settings,
            liveness_probe=liveness_probe,
            readiness_probe=readiness_probe,
            code_path=code_path,
            scoring_script=scoring_script,
            **kwargs,
        )

        self.resources = resources

    def _to_dict(self) -> Dict:
        res: dict = KubernetesOnlineDeploymentSchema(context={BASE_PATH_CONTEXT_KEY: "./"}).dump(self)
        return res

    # pylint: disable=arguments-differ
    def _to_rest_object(self, location: str) -> RestOnlineDeploymentData:  # type: ignore
        self._validate()
        code, environment, model = self._generate_dependencies()

        properties = RestKubernetesOnlineDeployment(
            code_configuration=code,
            environment_id=environment,
            model=model,
            model_mount_path=self.model_mount_path,
            scale_settings=self.scale_settings._to_rest_object() if self.scale_settings else None,
            properties=self.properties,
            description=self.description,
            environment_variables=self.environment_variables,
            app_insights_enabled=self.app_insights_enabled,
            request_settings=self.request_settings._to_rest_object() if self.request_settings else None,
            liveness_probe=self.liveness_probe._to_rest_object() if self.liveness_probe else None,
            readiness_probe=self.readiness_probe._to_rest_object() if self.readiness_probe else None,
            container_resource_requirements=self.resources._to_rest_object() if self.resources else None,
            instance_type=self.instance_type if self.instance_type else None,
            data_collector=self.data_collector._to_rest_object() if self.data_collector else None,
        )
        sku = RestSku(name="Default", capacity=self.instance_count)

        return RestOnlineDeploymentData(location=location, properties=properties, tags=self.tags, sku=sku)

    def _to_arm_resource_param(self, **kwargs: Any) -> Dict:
        rest_object = self._to_rest_object(**kwargs)
        properties = rest_object.properties
        sku = rest_object.sku
        tags = rest_object.tags

        return {
            self._arm_type: {
                ArmConstants.NAME: self.name,
                ArmConstants.PROPERTIES_PARAMETER_NAME: self._serialize.body(properties, "K8SOnlineDeployment"),
                ArmConstants.SKU: self._serialize.body(sku, "Sku"),
                ArmConstants.TAGS: tags,
            }
        }

    def _merge_with(self, other: Any) -> None:
        if other:
            super()._merge_with(other)
            if self.resources:
                self.resources._merge_with(other.resources)
            else:
                self.resources = other.resources

    def _validate(self) -> None:
        self._validate_name()

    @classmethod
    def _from_rest_object(cls, resource: RestOnlineDeploymentData) -> "KubernetesOnlineDeployment":
        deployment = resource.properties

        code_config = (
            CodeConfiguration(
                code=deployment.code_configuration.code_id,
                scoring_script=deployment.code_configuration.scoring_script,
            )
            if deployment.code_configuration
            else None
        )

        return KubernetesOnlineDeployment(
            id=resource.id,
            name=resource.name,
            tags=resource.tags,
            properties=deployment.properties,
            description=deployment.description,
            request_settings=OnlineRequestSettings._from_rest_object(deployment.request_settings),
            model=deployment.model,
            code_configuration=code_config,
            environment=deployment.environment_id,
            resources=ResourceRequirementsSettings._from_rest_object(deployment.container_resource_requirements),
            app_insights_enabled=deployment.app_insights_enabled,
            scale_settings=cast(
                Optional[Union[DefaultScaleSettings, TargetUtilizationScaleSettings]],
                OnlineScaleSettings._from_rest_object(deployment.scale_settings),
            ),
            liveness_probe=ProbeSettings._from_rest_object(deployment.liveness_probe),
            readiness_probe=ProbeSettings._from_rest_object(deployment.readiness_probe),
            environment_variables=deployment.environment_variables,
            endpoint_name=_parse_endpoint_name_from_deployment_id(resource.id),
            instance_count=resource.sku.capacity if resource.sku else None,
            instance_type=deployment.instance_type,
            data_collector=(
                DataCollector._from_rest_object(deployment.data_collector)
                if hasattr(deployment, "data_collector") and deployment.data_collector
                else None
            ),
            provisioning_state=deployment.provisioning_state if hasattr(deployment, "provisioning_state") else None,
        )


class ManagedOnlineDeployment(OnlineDeployment):
    """Managed Online endpoint deployment entity.

    :param name: Name of the deployment resource
    :type name: str
    :param endpoint_name: Name of the endpoint resource, defaults to None
    :type endpoint_name: typing.Optional[str]
    :param tags: Tag dictionary. Tags can be added, removed, and updated., defaults to None
    :type tags: typing.Optional[typing.Dict[str, typing.Any]]
    :param properties: The asset property dictionary, defaults to None
    :type properties: typing.Optional[typing.Dict[str, typing.Any]]
    :param description: Description of the resource, defaults to None
    :type description: typing.Optional[str]
    :param model: Model entity for the endpoint deployment, defaults to None
    :type model: typing.Optional[typing.Union[str, ~azure.ai.ml.entities.Model]]
    :param code_configuration: Code Configuration, defaults to None
    :type code_configuration: typing.Optional[~azure.ai.ml.entities.CodeConfiguration]
    :param environment: Environment entity for the endpoint deployment, defaults to None
    :type environment: typing.Optional[typing.Union[str, ~azure.ai.ml.entities.Environment]]
    :param app_insights_enabled: Is appinsights enabled, defaults to False
    :type app_insights_enabled: bool
    :param scale_settings: How the online deployment will scale, defaults to None
    :type scale_settings: typing.Optional[typing.Union[~azure.ai.ml.entities.DefaultScaleSettings
        , ~azure.ai.ml.entities.TargetUtilizationScaleSettings]]
    :param request_settings: Online Request Settings, defaults to None
    :type request_settings: typing.Optional[OnlineRequestSettings]
    :param liveness_probe: Liveness probe settings, defaults to None
    :type liveness_probe: typing.Optional[~azure.ai.ml.entities.ProbeSettings]
    :param readiness_probe: Readiness probe settings, defaults to None
    :type readiness_probe: typing.Optional[~azure.ai.ml.entities.ProbeSettings]
    :param environment_variables: Environment variables that will be set in deployment, defaults to None
    :type environment_variables: typing.Optional[typing.Dict[str, str]]
    :param instance_type: Azure compute sku, defaults to None
    :type instance_type: typing.Optional[str]
    :param instance_count: The instance count used for this deployment, defaults to None
    :type instance_count: typing.Optional[int]
    :param egress_public_network_access: Whether to restrict communication between a deployment and the
         Azure resources used to by the deployment. Allowed values are: "enabled", "disabled", defaults to None
    :type egress_public_network_access: typing.Optional[str]
    :param code_path: Equivalent to code_configuration.code, will be ignored if code_configuration is present
        , defaults to None
    :type code_path: typing.Optional[typing.Union[str, os.PathLike]]
    :param scoring_script_path: Equivalent to code_configuration.scoring_script, will be ignored if
        code_configuration is present, defaults to None
    :type scoring_script_path: typing.Optional[typing.Union[str, os.PathLike]]
    :param data_collector: Data collector, defaults to None
    :type data_collector: typing.Optional[typing.List[~azure.ai.ml.entities.DataCollector]]
    """

    def __init__(
        self,
        *,
        name: str,
        endpoint_name: Optional[str] = None,
        tags: Optional[Dict[str, typing.Any]] = None,
        properties: Optional[Dict[str, typing.Any]] = None,
        description: Optional[str] = None,
        model: Optional[Union[str, "Model"]] = None,
        code_configuration: Optional[CodeConfiguration] = None,
        environment: Optional[Union[str, "Environment"]] = None,
        app_insights_enabled: bool = False,
        scale_settings: Optional[Union[DefaultScaleSettings, TargetUtilizationScaleSettings]] = None,
        request_settings: Optional[OnlineRequestSettings] = None,
        liveness_probe: Optional[ProbeSettings] = None,
        readiness_probe: Optional[ProbeSettings] = None,
        environment_variables: Optional[Dict[str, str]] = None,
        instance_type: Optional[str] = None,
        instance_count: Optional[int] = None,
        egress_public_network_access: Optional[str] = None,
        code_path: Optional[Union[str, os.PathLike]] = None,  # promoted property from code_configuration.code
        scoring_script: Optional[
            Union[str, os.PathLike]
        ] = None,  # promoted property from code_configuration.scoring_script
        data_collector: Optional[DataCollector] = None,
        **kwargs: Any,
    ):
        kwargs["type"] = EndpointComputeType.MANAGED.value
        self.private_network_connection = kwargs.pop("private_network_connection", None)
        self.package_model = kwargs.pop("package_model", False)

        super(ManagedOnlineDeployment, self).__init__(
            name=name,
            endpoint_name=endpoint_name,
            tags=tags,
            properties=properties,
            description=description,
            model=model,
            code_configuration=code_configuration,
            environment=environment,
            environment_variables=environment_variables,
            app_insights_enabled=app_insights_enabled,
            scale_settings=scale_settings,
            request_settings=request_settings,
            liveness_probe=liveness_probe,
            readiness_probe=readiness_probe,
            instance_count=instance_count,
            instance_type=instance_type,
            code_path=code_path,
            scoring_script=scoring_script,
            data_collector=data_collector,
            **kwargs,
        )

        self.readiness_probe = readiness_probe
        self.egress_public_network_access = egress_public_network_access

    def _to_dict(self) -> Dict:
        res: dict = ManagedOnlineDeploymentSchema(context={BASE_PATH_CONTEXT_KEY: "./"}).dump(self)
        return res

    # pylint: disable=arguments-differ
    def _to_rest_object(self, location: str) -> RestOnlineDeploymentData:  # type: ignore
        self._validate()
        code, environment, model = self._generate_dependencies()
        properties = RestManagedOnlineDeployment(
            code_configuration=code,
            environment_id=environment,
            model=model,
            model_mount_path=self.model_mount_path,
            scale_settings=self.scale_settings._to_rest_object() if self.scale_settings else None,
            properties=self.properties,
            description=self.description,
            environment_variables=self.environment_variables,
            app_insights_enabled=self.app_insights_enabled,
            request_settings=self.request_settings._to_rest_object() if self.request_settings else None,
            liveness_probe=self.liveness_probe._to_rest_object() if self.liveness_probe else None,
            instance_type=self.instance_type,
            readiness_probe=self.readiness_probe._to_rest_object() if self.readiness_probe else None,
            data_collector=self.data_collector._to_rest_object() if self.data_collector else None,
        )
        # TODO: SKU name is defaulted to value "Default" since service side requires it.
        #  Should be removed once service side defaults it.
        sku = RestSku(name="Default", capacity=self.instance_count)

        # mfe is expecting private network connection to be in both the attribute level
        # as well as in the properties dictionary.
        if hasattr(self, "private_network_connection") and self.private_network_connection:
            properties.private_network_connection = self.private_network_connection
            properties.properties["private-network-connection"] = self.private_network_connection
        if hasattr(self, "egress_public_network_access") and self.egress_public_network_access:
            properties.egress_public_network_access = self.egress_public_network_access
        return RestOnlineDeploymentData(location=location, properties=properties, tags=self.tags, sku=sku)

    def _to_arm_resource_param(self, **kwargs: Any) -> Dict:
        rest_object = self._to_rest_object(**kwargs)
        properties = rest_object.properties
        sku = rest_object.sku
        tags = rest_object.tags

        return {
            self._arm_type: {
                ArmConstants.NAME: self.name,
                ArmConstants.PROPERTIES_PARAMETER_NAME: self._serialize.body(properties, "ManagedOnlineDeployment"),
                ArmConstants.SKU: self._serialize.body(sku, "Sku"),
                ArmConstants.TAGS: tags,
            }
        }

    @classmethod
    def _from_rest_object(cls, resource: RestOnlineDeploymentData) -> "ManagedOnlineDeployment":
        deployment = resource.properties

        code_config = (
            CodeConfiguration(
                code=deployment.code_configuration.code_id,
                scoring_script=deployment.code_configuration.scoring_script,
            )
            if deployment.code_configuration
            else None
        )

        return ManagedOnlineDeployment(
            id=resource.id,
            name=resource.name,
            tags=resource.tags,
            properties=deployment.properties,
            description=deployment.description,
            request_settings=OnlineRequestSettings._from_rest_object(deployment.request_settings),
            model=(deployment.model if deployment.model else None),
            code_configuration=code_config,
            environment=deployment.environment_id,
            app_insights_enabled=deployment.app_insights_enabled,
            scale_settings=OnlineScaleSettings._from_rest_object(deployment.scale_settings),  # type: ignore
            liveness_probe=ProbeSettings._from_rest_object(deployment.liveness_probe),
            environment_variables=deployment.environment_variables,
            readiness_probe=ProbeSettings._from_rest_object(deployment.readiness_probe),
            instance_type=deployment.instance_type,
            endpoint_name=_parse_endpoint_name_from_deployment_id(resource.id),
            instance_count=resource.sku.capacity,
            private_network_connection=(
                deployment.private_network_connection if hasattr(deployment, "private_network_connection") else None
            ),
            egress_public_network_access=deployment.egress_public_network_access,
            data_collector=(
                DataCollector._from_rest_object(deployment.data_collector)
                if hasattr(deployment, "data_collector") and deployment.data_collector
                else None
            ),
            provisioning_state=deployment.provisioning_state if hasattr(deployment, "provisioning_state") else None,
            creation_context=resource.system_data,
        )

    def _merge_with(self, other: Any) -> None:
        if other:
            super()._merge_with(other)
            self.instance_type = other.instance_type or self.instance_type

    def _validate(self) -> None:
        self._validate_name()
        self._validate_scale_settings()

    def _validate_scale_settings(self) -> None:
        if self.scale_settings:
            if not isinstance(self.scale_settings, DefaultScaleSettings):
                msg = "ManagedOnlineEndpoint supports DefaultScaleSettings only."
                raise ValidationException(
                    message=msg,
                    target=ErrorTarget.ONLINE_DEPLOYMENT,
                    no_personal_data_message=msg,
                    error_category=ErrorCategory.USER_ERROR,
                    error_type=ValidationErrorType.INVALID_VALUE,
                )