refactor: update all deprecated usage of Type

Author: lemorage

python/cocoindex/convert.py

@@ -8,7 +8,7 @@
 import uuid
 
 from enum import Enum
-from typing import Any, Callable, get_origin, Mapping, Type
+from typing import Any, Callable, get_origin, Mapping
 from .typing import (
     analyze_type_info,
     encode_enriched_type,
@@ -41,7 +41,7 @@ def encode_engine_value(value: Any) -> Any:
 def make_engine_value_decoder(
     field_path: list[str],
     src_type: dict[str, Any],
-    dst_annotation: Type[Any] | None,
+    dst_annotation: Any,
 ) -> Callable[[Any], Any]:
     """
     Make a decoder from an engine value to a Python value.
@@ -60,7 +60,7 @@ def make_engine_value_decoder(
     if (
         dst_annotation is None
         or dst_annotation is inspect.Parameter.empty
-        or get_origin(dst_annotation) is Any
+        or dst_annotation is Any
     ):
         if src_type_kind == "Struct" or src_type_kind in TABLE_TYPES:
             raise ValueError(

python/cocoindex/flow.py

@@ -18,7 +18,6 @@
     Generic,
     get_args,
     get_origin,
-    Type,
     NamedTuple,
     cast,
 )
@@ -786,14 +785,14 @@ async def _update_flow(name: str, fl: Flow) -> tuple[str, _engine.IndexUpdateInf
 
 
 def _get_data_slice_annotation_type(
-    data_slice_type: Type[DataSlice[T] | inspect._empty],
-) -> Type[T] | None:
+    data_slice_type: type[DataSlice[T] | inspect._empty],
+) -> type[T] | None:
     type_args = get_args(data_slice_type)
     if data_slice_type is inspect.Parameter.empty or data_slice_type is DataSlice:
         return None
     if get_origin(data_slice_type) != DataSlice or len(type_args) != 1:
         raise ValueError(f"Expect a DataSlice[T] type, but got {data_slice_type}")
-    return cast(Type[T] | None, type_args[0])
+    return cast(type[T] | None, type_args[0])
 
 
 _transform_flow_name_builder = _NameBuilder()
@@ -892,7 +891,7 @@ async def _build_flow_info_async(self) -> TransformFlowInfo:
         engine_return_type = (
             _data_slice_state(output).engine_data_slice.data_type().schema()
         )
-        python_return_type: Type[T] | None = _get_data_slice_annotation_type(
+        python_return_type: type[T] | None = _get_data_slice_annotation_type(
             sig.return_annotation
         )
         result_decoder = make_engine_value_decoder(
@@ -952,7 +951,7 @@ def _transform_flow_wrapper(fn: Callable[..., DataSlice[T]]) -> TransformFlow[T]
                 raise ValueError(
                     f"Parameter `{param_name}` is not a parameter can be passed by name"
                 )
-            value_type_annotation: Type[T] | None = _get_data_slice_annotation_type(
+            value_type_annotation: type[T] | None = _get_data_slice_annotation_type(
                 param.annotation
             )
             if value_type_annotation is None:

python/cocoindex/functions.py

@@ -1,6 +1,6 @@
 """All builtin functions."""
 
-from typing import Annotated, Any, Type, TYPE_CHECKING
+from typing import Annotated, Any, TYPE_CHECKING
 
 from .typing import Float32, Vector, TypeAttr
 from . import op, llm
@@ -47,13 +47,13 @@ class SentenceTransformerEmbedExecutor:
     spec: SentenceTransformerEmbed
     _model: "sentence_transformers.SentenceTransformer"
 
-    def analyze(self, text: Any) -> Type[Any]:
+    def analyze(self, text: Any) -> type:
         import sentence_transformers  # pylint: disable=import-outside-toplevel
 
         args = self.spec.args or {}
         self._model = sentence_transformers.SentenceTransformer(self.spec.model, **args)
         dim = self._model.get_sentence_embedding_dimension()
-        result: Type[Any] = Annotated[
+        result: type = Annotated[
             Vector[Float32, dim],  # type: ignore
             TypeAttr("cocoindex.io/vector_origin_text", text.analyzed_value),
         ]