fix bitwise OR for function typing in python 3.8

Author: VipulMascarenhas

ads/aqua/extension/common_ws_msg_handler.py

@@ -1,18 +1,25 @@
+#!/usr/bin/env python
+
+# Copyright (c) 2024 Oracle and/or its affiliates.
+# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
+
 import json
+from importlib import metadata
 from typing import List, Union
 
 from ads.aqua import ODSC_MODEL_COMPARTMENT_OCID, fetch_service_compartment
 from ads.aqua.common.decorator import handle_exceptions
 from ads.aqua.common.errors import AquaResourceAccessError
 from ads.aqua.common.utils import known_realm
 from ads.aqua.extension.aqua_ws_msg_handler import AquaWSMsgHandler
-from ads.aqua.extension.models.ws_models import RequestResponseType, AdsVersionResponse, AdsVersionRequest, \
-    CompatibilityCheckResponse
-from importlib import metadata
+from ads.aqua.extension.models.ws_models import (
+    AdsVersionResponse,
+    CompatibilityCheckResponse,
+    RequestResponseType,
+)
 
 
 class AquaCommonWsMsgHandler(AquaWSMsgHandler):
-
     @staticmethod
     def get_message_types() -> List[RequestResponseType]:
         return [RequestResponseType.AdsVersion, RequestResponseType.CompatibilityCheck]
@@ -21,25 +28,30 @@ def __init__(self, message: Union[str, bytes]):
         super().__init__(message)
 
     @handle_exceptions
-    def process(self) -> AdsVersionResponse | CompatibilityCheckResponse:
+    def process(self) -> Union[AdsVersionResponse, CompatibilityCheckResponse]:
         request = json.loads(self.message)
-        if request.get('kind') == 'AdsVersion':
+        if request.get("kind") == "AdsVersion":
             version = metadata.version("oracle_ads")
             response = AdsVersionResponse(
                 message_id=request.get("message_id"),
                 kind=RequestResponseType.AdsVersion,
-                data=version)
+                data=version,
+            )
             return response
-        if request.get('kind') == 'CompatibilityCheck':
+        if request.get("kind") == "CompatibilityCheck":
             if ODSC_MODEL_COMPARTMENT_OCID or fetch_service_compartment():
-                return CompatibilityCheckResponse(message_id=request.get("message_id"),
-                                                  kind=RequestResponseType.CompatibilityCheck,
-                                                  data={'status': 'ok'})
+                return CompatibilityCheckResponse(
+                    message_id=request.get("message_id"),
+                    kind=RequestResponseType.CompatibilityCheck,
+                    data={"status": "ok"},
+                )
             elif known_realm():
-                return CompatibilityCheckResponse(message_id=request.get("message_id"),
-                                                  kind=RequestResponseType.CompatibilityCheck,
-                                                  data={'status': 'compatible'})
+                return CompatibilityCheckResponse(
+                    message_id=request.get("message_id"),
+                    kind=RequestResponseType.CompatibilityCheck,
+                    data={"status": "compatible"},
+                )
             else:
                 raise AquaResourceAccessError(
-                    f"The AI Quick actions extension is not compatible in the given region."
+                    "The AI Quick actions extension is not compatible in the given region."
                 )

ads/aqua/extension/deployment_ws_msg_handler.py

@@ -1,25 +1,35 @@
+#!/usr/bin/env python
+
+# Copyright (c) 2024 Oracle and/or its affiliates.
+# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
+
 import json
 from typing import List, Union
 
 from ads.aqua.common.decorator import handle_exceptions
 from ads.aqua.extension.aqua_ws_msg_handler import AquaWSMsgHandler
-from ads.aqua.extension.models.ws_models import RequestResponseType, ListDeploymentResponse, ListDeploymentRequest, \
-    ModelDeploymentDetailsResponse
+from ads.aqua.extension.models.ws_models import (
+    ListDeploymentResponse,
+    ModelDeploymentDetailsResponse,
+    RequestResponseType,
+)
 from ads.aqua.modeldeployment import AquaDeploymentApp
 from ads.config import COMPARTMENT_OCID
 
 
 class AquaDeploymentWSMsgHandler(AquaWSMsgHandler):
-
     def __init__(self, message: Union[str, bytes]):
         super().__init__(message)
 
     @staticmethod
     def get_message_types() -> List[RequestResponseType]:
-        return [RequestResponseType.ListDeployments, RequestResponseType.DeploymentDetails]
+        return [
+            RequestResponseType.ListDeployments,
+            RequestResponseType.DeploymentDetails,
+        ]
 
     @handle_exceptions
-    def process(self) -> ListDeploymentResponse | ModelDeploymentDetailsResponse:
+    def process(self) -> Union[ListDeploymentResponse, ModelDeploymentDetailsResponse]:
         request = json.loads(self.message)
         if request.get("kind") == "ListDeployments":
             deployment_list = AquaDeploymentApp().list(
@@ -33,8 +43,12 @@ def process(self) -> ListDeploymentResponse | ModelDeploymentDetailsResponse:
             )
             return response
         elif request.get("kind") == "DeploymentDetails":
-            deployment_details = AquaDeploymentApp().get(request.get("model_deployment_id"))
-            response = ModelDeploymentDetailsResponse(message_id=request.get("message_id"),
-                                                      kind=RequestResponseType.DeploymentDetails,
-                                                      data=deployment_details)
+            deployment_details = AquaDeploymentApp().get(
+                request.get("model_deployment_id")
+            )
+            response = ModelDeploymentDetailsResponse(
+                message_id=request.get("message_id"),
+                kind=RequestResponseType.DeploymentDetails,
+                data=deployment_details,
+            )
             return response

ads/aqua/extension/evaluation_ws_msg_handler.py

@@ -29,7 +29,7 @@ def __init__(self, message: Union[str, bytes]):
         super().__init__(message)
 
     @handle_exceptions
-    def process(self) -> ListEvaluationsResponse | EvaluationDetailsResponse:
+    def process(self) -> Union[ListEvaluationsResponse, EvaluationDetailsResponse]:
         request = json.loads(self.message)
         if request["kind"] == "ListEvaluations":
             return self.list_evaluations(request)

ads/aqua/extension/models_ws_msg_handler.py

@@ -1,41 +1,49 @@
+#!/usr/bin/env python
+
+# Copyright (c) 2024 Oracle and/or its affiliates.
+# Licensed under the Universal Permissive License v 1.0 as shown at https://oss.oracle.com/licenses/upl/
+
 import json
 from typing import List, Union
 
 from ads.aqua.common.decorator import handle_exceptions
 from ads.aqua.extension.aqua_ws_msg_handler import AquaWSMsgHandler
-from ads.aqua.extension.models.ws_models import RequestResponseType, ListModelsResponse, ListModelsRequest, \
-    ModelDetailsResponse
+from ads.aqua.extension.models.ws_models import (
+    ListModelsResponse,
+    ModelDetailsResponse,
+    RequestResponseType,
+)
 from ads.aqua.model import AquaModelApp
 
 
 class AquaModelWSMsgHandler(AquaWSMsgHandler):
-
     def __init__(self, message: Union[str, bytes]):
         super().__init__(message)
 
     @staticmethod
     def get_message_types() -> List[RequestResponseType]:
-        return [RequestResponseType.ListModels,RequestResponseType.ModelDetails]
+        return [RequestResponseType.ListModels, RequestResponseType.ModelDetails]
 
     @handle_exceptions
-    def process(self) -> ListModelsResponse | ModelDetailsResponse:
+    def process(self) -> Union[ListModelsResponse, ModelDetailsResponse]:
         request = json.loads(self.message)
-        if request.get('kind') == 'ListModels':
+        if request.get("kind") == "ListModels":
             models_list = AquaModelApp().list(
                 compartment_id=request.get("compartment_id"),
                 project_id=request.get("project_id"),
-                model_type=request.get("model_type")
+                model_type=request.get("model_type"),
             )
             response = ListModelsResponse(
                 message_id=request.get("message_id"),
                 kind=RequestResponseType.ListModels,
                 data=models_list,
             )
             return response
-        elif request.get('kind') == 'ModelDetails':
-            model_id=request.get("model_id")
-            response=AquaModelApp().get(model_id)
-            return ModelDetailsResponse(message_id=request.get("message_id"),
-                                        kind=RequestResponseType.ModelDetails,
-                                        data=response)
-
+        elif request.get("kind") == "ModelDetails":
+            model_id = request.get("model_id")
+            response = AquaModelApp().get(model_id)
+            return ModelDetailsResponse(
+                message_id=request.get("message_id"),
+                kind=RequestResponseType.ModelDetails,
+                data=response,
+            )