Fix marshmallow 4.x context and formatting issues: update load_from_dict function and apply black formatting

Co-authored-by: kshitij-microsoft <[email protected]>

Author: Copilot

=4.0.0,

@@ -15,18 +15,30 @@
     _get_cloud_details,
     _resource_to_scopes,
 )
-from azure.ai.ml._utils._arm_id_utils import AzureResourceId, get_arm_id_object_from_id
+from azure.ai.ml._utils._arm_id_utils import (
+    AzureResourceId,
+    get_arm_id_object_from_id,
+)
 from azure.ai.ml._utils._logger_utils import initialize_logger_info
 from azure.ai.ml._utils.utils import from_iso_duration_format_min_sec
-from azure.ai.ml._vendor.azure_resources._resource_management_client import ResourceManagementClient
-from azure.ai.ml._vendor.azure_resources.models import Deployment, DeploymentProperties
+from azure.ai.ml._vendor.azure_resources._resource_management_client import (
+    ResourceManagementClient,
+)
+from azure.ai.ml._vendor.azure_resources.models import (
+    Deployment,
+    DeploymentProperties,
+)
 from azure.ai.ml.constants._common import (
     ENDPOINT_DEPLOYMENT_START_MSG,
     ArmConstants,
     LROConfigurations,
     OperationStatus,
 )
-from azure.ai.ml.exceptions import ErrorCategory, ErrorTarget, ValidationException
+from azure.ai.ml.exceptions import (
+    ErrorCategory,
+    ErrorTarget,
+    ValidationException,
+)
 from azure.core.credentials import TokenCredential
 from azure.core.polling import LROPoller
 
@@ -65,7 +77,9 @@ def __init__(
         self._deployments_client = self._client.deployments
         self._deployment_tracking = []
         self._lock = None  # To allow only one deployment to print
-        self._printed_set = set()  # To prevent already printed deployment from re using the console
+        self._printed_set = (
+            set()
+        )  # To prevent already printed deployment from re using the console
         self._resources_being_deployed = {}
 
     def deploy_resource(
@@ -88,14 +102,17 @@ def deploy_resource(
         try:
             poller = self._get_poller(template=template, parameters=parameters)
             module_logger.info(
-                "The deployment request %s was accepted. ARM deployment URI for reference: \n", self._deployment_name
-            )
-            endpoint_deployment_start_message = ENDPOINT_DEPLOYMENT_START_MSG.format(
-                _get_azure_portal_id_from_metadata(),
-                self._subscription_id,
-                self._resource_group_name,
+                "The deployment request %s was accepted. ARM deployment URI for reference: \n",
                 self._deployment_name,
             )
+            endpoint_deployment_start_message = (
+                ENDPOINT_DEPLOYMENT_START_MSG.format(
+                    _get_azure_portal_id_from_metadata(),
+                    self._subscription_id,
+                    self._resource_group_name,
+                    self._deployment_name,
+                )
+            )
             module_logger.info(endpoint_deployment_start_message)
             if wait:
                 try:
@@ -120,20 +137,32 @@ def deploy_resource(
             else:
                 return poller
         except Exception as ex:
-            module_logger.debug("Polling hit the exception: %s", type(ex).__name__)
+            module_logger.debug(
+                "Polling hit the exception: %s", type(ex).__name__
+            )
             raise ex
 
         if error is not None:
             error_msg = f"Unable to create resource. \n {error}\n"
             module_logger.error(error_msg)
             raise error
         if len(resources_being_deployed) > 1 and total_duration:
-            module_logger.info("Total time : %s\n", from_iso_duration_format_min_sec(total_duration))
+            module_logger.info(
+                "Total time : %s\n",
+                from_iso_duration_format_min_sec(total_duration),
+            )
         return None
 
-    def _get_poller(self, template: str, parameters: Optional[Dict] = None, wait: bool = True) -> None:
+    def _get_poller(
+        self,
+        template: str,
+        parameters: Optional[Dict] = None,
+        wait: bool = True,
+    ) -> None:
         # deploy the template
-        properties = DeploymentProperties(template=template, parameters=parameters, mode="incremental")
+        properties = DeploymentProperties(
+            template=template, parameters=parameters, mode="incremental"
+        )
         return self._deployments_client.begin_create_or_update(
             resource_group_name=self._resource_group_name,
             deployment_name=self._deployment_name,
@@ -165,36 +194,52 @@ def _check_deployment_status(self) -> None:
 
             arm_id_obj = get_arm_id_object_from_id(target_resource.id)
 
-            resource_name = (
-                f"{arm_id_obj.asset_name} {arm_id_obj.asset_version if hasattr(arm_id_obj,'asset_version') else ''}"
-            )
+            resource_name = f"{arm_id_obj.asset_name} {arm_id_obj.asset_version if hasattr(arm_id_obj,'asset_version') else ''}"
             # do swap on asset_type to avoid collision with workspaces asset_type in arm id
             if isinstance(arm_id_obj, AzureResourceId):
                 arm_id_obj.asset_type = (
                     arm_id_obj.asset_type
-                    if not arm_id_obj.provider_namespace_with_type == "OperationalInsightsworkspaces"
+                    if not arm_id_obj.provider_namespace_with_type
+                    == "OperationalInsightsworkspaces"
                     else "LogAnalytics"
                 )
-            deployment_message = deployment_message_mapping[arm_id_obj.asset_type].format(f"{resource_name} ")
-            if target_resource.resource_name not in self._resources_being_deployed:
-                self._resources_being_deployed[target_resource.resource_name] = (
+            deployment_message = deployment_message_mapping[
+                arm_id_obj.asset_type
+            ].format(f"{resource_name} ")
+            if (
+                target_resource.resource_name
+                not in self._resources_being_deployed
+            ):
+                self._resources_being_deployed[
+                    target_resource.resource_name
+                ] = (
                     deployment_message,
                     None,
                 )
 
             if (
                 properties.provisioning_state
-                and (not self._lock or self._lock == target_resource.resource_name)
+                and (
+                    not self._lock
+                    or self._lock == target_resource.resource_name
+                )
                 and target_resource.resource_name not in self._printed_set
             ):
-                status_in_resource_dict = self._resources_being_deployed[target_resource.resource_name][1]
+                status_in_resource_dict = self._resources_being_deployed[
+                    target_resource.resource_name
+                ][1]
                 module_logger.debug(
-                    ("\n LOCK STATUS :  %s,  Status in the resources dict : %s ,  Already in printed set: %s\n"),
+                    (
+                        "\n LOCK STATUS :  %s,  Status in the resources dict : %s ,  Already in printed set: %s\n"
+                    ),
                     self._lock,
                     status_in_resource_dict,
                     self._printed_set,
                 )
-                module_logger.debug("Locking with the deployment : %s\n\n", target_resource.resource_name)
+                module_logger.debug(
+                    "Locking with the deployment : %s\n\n",
+                    target_resource.resource_name,
+                )
                 self._lock = target_resource.resource_name
                 provisioning_state = properties.provisioning_state
                 request_id = properties.service_request_id
@@ -206,12 +251,16 @@ def _check_deployment_status(self) -> None:
                 if resource_name not in self._resources_being_deployed:
                     resource_type, previous_state = resource_name, None
                 else:
-                    resource_type, previous_state = self._resources_being_deployed[resource_name]
+                    resource_type, previous_state = (
+                        self._resources_being_deployed[resource_name]
+                    )
 
                 duration = properties.duration
                 # duration comes in format: "PT1M56.3454108S"
                 try:
-                    duration_in_min_sec = from_iso_duration_format_min_sec(duration)
+                    duration_in_min_sec = from_iso_duration_format_min_sec(
+                        duration
+                    )
                 except Exception:  # pylint: disable=W0718
                     duration_in_min_sec = ""
 
@@ -220,7 +269,10 @@ def _check_deployment_status(self) -> None:
                     provisioning_state,
                 )
 
-                if provisioning_state == OperationStatus.FAILED and previous_state != OperationStatus.FAILED:
+                if (
+                    provisioning_state == OperationStatus.FAILED
+                    and previous_state != OperationStatus.FAILED
+                ):
                     status_code = properties.status_code
                     status_message = properties.status_message
                     module_logger.debug(
@@ -237,11 +289,18 @@ def _check_deployment_status(self) -> None:
                     )
                     module_logger.debug(
                         "More details: %s\n",
-                        status_message.error.details[0].message if status_message.error.details else None,
+                        (
+                            status_message.error.details[0].message
+                            if status_message.error.details
+                            else None
+                        ),
                     )
                     # self._lock = None
                 # First time we're seeing this so let the user know it's being deployed
-                elif properties.provisioning_state == OperationStatus.RUNNING and previous_state is None:
+                elif (
+                    properties.provisioning_state == OperationStatus.RUNNING
+                    and previous_state is None
+                ):
                     module_logger.info("%s ", resource_type)
                 elif (
                     properties.provisioning_state == OperationStatus.RUNNING
@@ -251,11 +310,19 @@ def _check_deployment_status(self) -> None:
                 # If the provisioning has already succeeded but we hadn't seen it Running before
                 # (really quick deployment - so probably never happening) let user know resource
                 # is being deployed and then let user know it has been deployed
-                elif properties.provisioning_state == OperationStatus.SUCCEEDED and previous_state is None:
-                    module_logger.info("%s  Done (%s)\n", resource_type, duration_in_min_sec)
+                elif (
+                    properties.provisioning_state == OperationStatus.SUCCEEDED
+                    and previous_state is None
+                ):
+                    module_logger.info(
+                        "%s  Done (%s)\n", resource_type, duration_in_min_sec
+                    )
                     self._lock = None
                     self._printed_set.add(resource_name)
-                    module_logger.debug("Releasing lock for deployment: %s\n\n", target_resource.resource_name)
+                    module_logger.debug(
+                        "Releasing lock for deployment: %s\n\n",
+                        target_resource.resource_name,
+                    )
                 # Finally, deployment has succeeded and was previously running, so mark it as finished
                 elif (
                     properties.provisioning_state == OperationStatus.SUCCEEDED
@@ -264,4 +331,7 @@ def _check_deployment_status(self) -> None:
                     module_logger.info("  Done (%s)\n", duration_in_min_sec)
                     self._lock = None
                     self._printed_set.add(resource_name)
-                    module_logger.debug("Releasing lock for deployment: %s\n\n", target_resource.resource_name)
+                    module_logger.debug(
+                        "Releasing lock for deployment: %s\n\n",
+                        target_resource.resource_name,
+                    )

sdk/ml/azure-ai-ml/azure/ai/ml/_arm_deployments/arm_deployment_executor.py

@@ -55,7 +55,17 @@
 
 def get_template(resource_type: str) -> Dict[str, Any]:
     if resource_type not in template_mapping:
-        msg = "can't find the template for the resource {}".format(resource_type)
-        raise ValidationException(message=msg, no_personal_data_message=msg, target=ErrorTarget.ARM_RESOURCE)
-    template_path = path.join(path.dirname(__file__), "arm_templates", template_mapping[resource_type])
+        msg = "can't find the template for the resource {}".format(
+            resource_type
+        )
+        raise ValidationException(
+            message=msg,
+            no_personal_data_message=msg,
+            target=ErrorTarget.ARM_RESOURCE,
+        )
+    template_path = path.join(
+        path.dirname(__file__),
+        "arm_templates",
+        template_mapping[resource_type],
+    )
     return load_json(file_path=template_path)