Eliminate old-fashioned type annotations in index_service

Author: nadove-ucsc

src/azul/indexer/index_service.py

@@ -17,11 +17,7 @@
 )
 from typing import (
     Any,
-    MutableSet,
-    Optional,
     TYPE_CHECKING,
-    Type,
-    Union,
     cast,
 )
 
@@ -267,7 +263,7 @@ def transform(self,
                   partition: BundlePartition = BundlePartition.root,
                   *,
                   delete: bool,
-                  ) -> Union[list[BundlePartition], tuple[list[Contribution], list[Replica]]]:
+                  ) -> list[BundlePartition] | tuple[list[Contribution], list[Replica]]:
         """
         Return a list of contributions and a list of replicas for the entities
         in the given partition of the specified bundle, or a set of divisions of
@@ -353,7 +349,7 @@ def stringify(value: AnyJSON) -> AnyJSON:
             )
 
         def setify(value: CompositeJSON
-                   ) -> Union[set[tuple[str, AnyJSON]], set[AnyJSON]]:
+                   ) -> set[tuple[str, AnyJSON]] | set[AnyJSON]:
             value = freeze(value)
             return set(
                 value.items()
@@ -578,7 +574,7 @@ def _read_contributions(self,
                             ) -> list[CataloguedContribution]:
         es_client = ESClientFactory.get()
 
-        entity_ids_by_index: dict[str, MutableSet[str]] = defaultdict(set)
+        entity_ids_by_index: dict[str, set[str]] = defaultdict(set)
         for entity in tallies.keys():
             index = str(IndexName.create(catalog=entity.catalog,
                                          qualifier=entity.entity_type,
@@ -696,7 +692,7 @@ def _aggregate(self,
             )
 
         # Create lookup for transformer by entity type
-        transformers: dict[tuple[CatalogName, str], Type[Transformer]] = {
+        transformers: dict[tuple[CatalogName, str], type[Transformer]] = {
             (catalog, transformer_cls.entity_type()): transformer_cls
             for catalog in config.catalogs
             for transformer_cls in self.transformer_types(catalog)
@@ -734,7 +730,7 @@ def _aggregate(self,
         return aggregates
 
     def _aggregate_entity(self,
-                          transformer: Type[Transformer],
+                          transformer: type[Transformer],
                           contributions: list[Contribution]
                           ) -> JSON:
         contents = self._reconcile(transformer, contributions)
@@ -756,7 +752,7 @@ def _aggregate_entity(self,
         return aggregate_contents
 
     def _reconcile(self,
-                   transformer: Type[Transformer],
+                   transformer: type[Transformer],
                    contributions: Sequence[Contribution],
                    ) -> Mapping[EntityType, Entities]:
         """
@@ -791,7 +787,7 @@ def _reconcile(self,
 
     def _create_writer(self,
                        doc_type: DocumentType,
-                       catalog: Optional[CatalogName]
+                       catalog: CatalogName | None
                        ) -> 'IndexWriter':
         # We allow one conflict retry in the case of duplicate notifications and
         # switch from 'add' to 'update'. After that, there should be no
@@ -814,9 +810,9 @@ def _create_writer(self,
 class IndexWriter:
 
     def __init__(self,
-                 catalog: Optional[CatalogName],
+                 catalog: CatalogName | None,
                  field_types: CataloguedFieldTypes,
-                 refresh: Union[bool, str],
+                 refresh: bool | str,
                  conflict_retry_limit: int,
                  error_retry_limit: int) -> None:
         """
@@ -843,7 +839,7 @@ def __init__(self,
         self.es_client = ESClientFactory.get()
         self.errors: dict[DocumentCoordinates, int] = defaultdict(int)
         self.conflicts: dict[DocumentCoordinates, int] = defaultdict(int)
-        self.retries: Optional[MutableSet[DocumentCoordinates]] = None
+        self.retries: set[DocumentCoordinates] | None = None
 
     bulk_threshold = 32
 
@@ -945,7 +941,7 @@ def _on_success(self, doc: Document):
         else:
             log.debug('Successfully wrote %s.', coordinates)
 
-    def _on_error(self, doc: Document, e: Union[Exception, JSON]):
+    def _on_error(self, doc: Document, e: Exception | JSON):
         self.errors[doc.coordinates] += 1
         if self.error_retry_limit is None or self.errors[doc.coordinates] <= self.error_retry_limit:
             action = 'retrying'
@@ -956,7 +952,7 @@ def _on_error(self, doc: Document, e: Union[Exception, JSON]):
                     doc.coordinates, e, self.errors[doc.coordinates], action,
                     exc_info=isinstance(e, Exception))
 
-    def _on_conflict(self, doc: Document, e: Union[Exception, JSON]):
+    def _on_conflict(self, doc: Document, e: Exception | JSON):
         self.conflicts[doc.coordinates] += 1
         self.errors.pop(doc.coordinates, None)  # a conflict resets the error count
         if self.conflict_retry_limit is None or self.conflicts[doc.coordinates] <= self.conflict_retry_limit: