typing v1 (#409)

libs/infinity_emb/infinity_emb/engine.py

@@ -2,7 +2,7 @@
 # Copyright (c) 2023-now michaelfeilfeil
 
 from asyncio import Semaphore
-from typing import Iterable, Iterator, List, Optional, Set, Union
+from typing import Iterable, Iterator, Optional, Union
 
 from infinity_emb.args import EngineArgs
 
@@ -122,7 +122,7 @@ def is_running(self) -> bool:
         return self.running
 
     @property
-    def capabilities(self) -> Set[ModelCapabilites]:
+    def capabilities(self) -> set[ModelCapabilites]:
         return self._model.capabilities
 
     @property
@@ -214,7 +214,7 @@ async def classify(
         return scores, usage
 
     async def image_embed(
-        self, *, images: List[Union[str, "ImageClassType", bytes]]
+        self, *, images: list[Union[str, "ImageClassType", bytes]]
     ) -> tuple[list["EmbeddingReturnType"], int]:
         """embed multiple images
 
@@ -237,7 +237,7 @@ async def image_embed(
         return embeddings, usage
 
     async def audio_embed(
-        self, *, audios: List[Union[str, bytes]]
+        self, *, audios: list[Union[str, bytes]]
     ) -> tuple[list["EmbeddingReturnType"], int]:
         """embed multiple audios
 
@@ -383,7 +383,7 @@ async def classify(
         return await self[model].classify(sentences=sentences, raw_scores=raw_scores)
 
     async def image_embed(
-        self, *, model: str, images: List[Union[str, "ImageClassType"]]
+        self, *, model: str, images: list[Union[str, "ImageClassType"]]
     ) -> tuple[list["EmbeddingReturnType"], int]:
         """embed multiple images
 

libs/infinity_emb/infinity_emb/fastapi_schemas/data_uri.py

@@ -5,7 +5,7 @@
 from base64 import b64decode as decode64
 from base64 import b64encode as encode64
 from dataclasses import dataclass
-from typing import Any, Dict, MutableMapping, Optional, Tuple, TypeVar, Union
+from typing import Any, MutableMapping, Optional, TypeVar, Union
 
 if sys.version_info >= (3, 11):
     from typing import Self
@@ -162,7 +162,7 @@ def is_valid(self) -> bool:
     @property
     def _parse(
         self,
-    ) -> Tuple[Optional[str], Optional[str], Optional[str], bool, bytes]:
+    ) -> tuple[Optional[str], Optional[str], Optional[str], bool, bytes]:
         match = _DATA_URI_RE.match(self)
         if match is None:
             raise InvalidDataURI("Not a valid data URI: %r" % self)
@@ -228,7 +228,7 @@ def __get_pydantic_json_schema__(
         return json_schema
 
     @classmethod
-    def __modify_schema__(cls, field_schema: Dict[str, Any]) -> None:
+    def __modify_schema__(cls, field_schema: dict[str, Any]) -> None:
         # __modify_schema__ should mutate the dict it receives in place,
         # the returned value will be ignored
         field_schema.update(

libs/infinity_emb/infinity_emb/inference/batch_handler.py

@@ -9,7 +9,7 @@
 import time
 from concurrent.futures import ThreadPoolExecutor
 from queue import Queue
-from typing import Any, List, Optional, Sequence, Set, Union
+from typing import Any, Optional, Sequence, Union
 
 import numpy as np
 
@@ -228,7 +228,7 @@ async def classify(
     async def image_embed(
         self,
         *,
-        images: List[Union[str, "ImageClassType", bytes]],
+        images: list[Union[str, "ImageClassType", bytes]],
     ) -> tuple[list["EmbeddingReturnType"], int]:
         """Schedule a images and sentences to be embedded. Awaits until embedded.
 
@@ -257,7 +257,7 @@ async def image_embed(
     async def audio_embed(
         self,
         *,
-        audios: List[Union[str, bytes]],
+        audios: list[Union[str, bytes]],
     ) -> tuple[list["EmbeddingReturnType"], int]:
         """Schedule audios and sentences to be embedded. Awaits until embedded.
 
@@ -310,7 +310,7 @@ async def _schedule(
         return result, usage
 
     @property
-    def capabilities(self) -> Set[ModelCapabilites]:
+    def capabilities(self) -> set[ModelCapabilites]:
         # TODO: try to remove inheritance here and return upon init.
         return self.model_worker.capabilities
 
@@ -447,7 +447,7 @@ def spawn(self):
         self._threadpool.submit(self._postprocess_batch)
 
     @property
-    def capabilities(self) -> Set[ModelCapabilites]:
+    def capabilities(self) -> set[ModelCapabilites]:
         return self._model.capabilities
 
     def tokenize_lengths(self, *args, **kwargs):

libs/infinity_emb/infinity_emb/infinity_server.py

@@ -7,7 +7,7 @@
 import sys
 import time
 from contextlib import asynccontextmanager
-from typing import Any, List, Optional, Union
+from typing import Any, Optional, Union
 
 import infinity_emb
 from infinity_emb._optional_imports import CHECK_TYPER, CHECK_UVICORN
@@ -204,8 +204,8 @@ def _resolve_engine(model: str) -> "AsyncEmbeddingEngine":
         return engine
 
     def _resolve_mixed_input(
-        inputs: Union[DataURIorURL, List[DataURIorURL]]
-    ) -> List[Union[str, bytes]]:
+        inputs: Union[DataURIorURL, list[DataURIorURL]]
+    ) -> list[Union[str, bytes]]:
         if hasattr(inputs, "host"):
             # if it is a single url
             urls_or_bytes: list[Union[str, bytes]] = [str(inputs)]

libs/infinity_emb/infinity_emb/sync_engine.py

@@ -12,7 +12,6 @@
     Awaitable,
     Callable,
     Iterator,
-    List,
     Optional,
     TypeVar,
     Union,
@@ -213,7 +212,7 @@ def classify(self, *, model: str, sentences: list[str], raw_scores: bool = False
         )
 
     @add_start_docstrings(AsyncEngineArray.image_embed.__doc__)
-    def image_embed(self, *, model: str, images: List[Union[str, ImageClassType]]):
+    def image_embed(self, *, model: str, images: list[Union[str, ImageClassType]]):
         """sync interface of AsyncEngineArray"""
         return self.async_run(
             self.async_engine_array.image_embed, model=model, images=images

libs/infinity_emb/infinity_emb/transformer/abstract.py

@@ -4,7 +4,7 @@
 import random
 from abc import ABC, abstractmethod
 from time import perf_counter
-from typing import TYPE_CHECKING, Any, Set, Union
+from typing import TYPE_CHECKING, Any, Union
 
 from infinity_emb._optional_imports import CHECK_PIL  # , CHECK_SOUNDFILE
 from infinity_emb.primitives import (
@@ -42,7 +42,7 @@
 
 
 class BaseTransformer(ABC):  # Inherit from ABC(Abstract base class)
-    capabilities: Set[ModelCapabilites] = set()
+    capabilities: set[ModelCapabilites] = set()
     engine_args: "EngineArgs"
 
     @abstractmethod  # Decorator to define an abstract method

libs/infinity_emb/infinity_emb/transformer/vision/utils.py

@@ -3,7 +3,7 @@
 
 import asyncio
 import io
-from typing import List, Union
+from typing import Union
 
 from infinity_emb._optional_imports import CHECK_AIOHTTP, CHECK_PIL
 from infinity_emb.primitives import (
@@ -76,8 +76,8 @@ async def resolve_image(
 
 
 async def resolve_images(
-    images: List[Union[str, "ImageClassType", bytes]]
-) -> List[ImageSingle]:
+    images: list[Union[str, "ImageClassType", bytes]]
+) -> list[ImageSingle]:
     """Resolve images from URLs or ImageClassType Objects using multithreading."""
     # TODO: improve parallel requests, safety, error handling
     CHECK_AIOHTTP.mark_required()