Fixed CI/CD error

Author: ryota-cloud

metadata-ingestion/src/datahub/ingestion/source/vertexai.py

@@ -1,7 +1,7 @@
 import dataclasses
 import logging
 from datetime import datetime
-from typing import Any, Iterable, List, Optional, TypeVar
+from typing import Any, Dict, Iterable, List, Optional, TypeVar
 
 from google.api_core.exceptions import GoogleAPICallError
 from google.cloud import aiplatform
@@ -140,8 +140,8 @@ def __init__(self, ctx: PipelineContext, config: VertexAIConfig):
             project=config.project_id, location=config.region, credentials=credentials
         )
         self.client = aiplatform
-        self.endpoints: Optional[dict[str, List[Endpoint]]] = None
-        self.datasets: Optional[dict[str, VertexAiResourceNoun]] = None
+        self.endpoints: Optional[Dict[str, List[Endpoint]]] = None
+        self.datasets: Optional[Dict[str, VertexAiResourceNoun]] = None
 
     def get_report(self) -> SourceReport:
         return self.report
@@ -395,7 +395,7 @@ def _search_dataset(self, dataset_id: str) -> Optional[VertexAiResourceNoun]:
         ]
 
         if self.datasets is None:
-            self.datasets = dict[str, VertexAiResourceNoun]()
+            self.datasets: Dict[str, VertexAiResourceNoun] = {}
 
             for dtype in dataset_types:
                 dataset_class = getattr(self.client.datasets, dtype)
@@ -633,7 +633,7 @@ def _search_endpoint(self, model: Model) -> List[Endpoint]:
         Search for an endpoint associated with the model.
         """
         if self.endpoints is None:
-            endpoint_dict = dict[str, List[Endpoint]]()
+            endpoint_dict:Dict[str, List[Endpoint]] = {}
             for endpoint in self.client.Endpoint.list():
                 for resource in endpoint.list_models():
                     if resource.model not in endpoint_dict: