Unify type of params to Dict[str, Any] or DefaultDict[str, Dict

NatLibFi · juhoinkinen · Jun 5, 2023 · May 23, 2023 · May 23, 2023 · May 23, 2023
commit 33cdcf2a53c1d7bd0d74da2c59ca53369bc4aff1
diff --git a/annif/backend/backend.py b/annif/backend/backend.py
@@ -5,7 +5,7 @@
 import os.path
 from datetime import datetime, timezone
 from glob import glob
-from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional
 
 from annif import logger
 from annif.suggestion import SuggestionBatch
@@ -24,7 +24,7 @@ class AnnifBackend(metaclass=abc.ABCMeta):
     DEFAULT_PARAMETERS = {"limit": 100}
 
     def __init__(
-        self, backend_id: str, config_params: Any, project: AnnifProject
+        self, backend_id: str, config_params: Dict[str, Any], project: AnnifProject
     ) -> None:
         """Initialize backend with specific parameters. The
         parameters are a dict. Keys and values depend on the specific
@@ -34,7 +34,7 @@ def __init__(
         self.project = project
         self.datadir = project.datadir
 
-    def default_params(self) -> Dict[str, Union[str, bool, int]]:
+    def default_params(self) -> Dict[str, Any]:
         return self.DEFAULT_PARAMETERS
 
     @property
@@ -61,7 +61,7 @@ def modification_time(self) -> Optional[datetime.datetime]:
 
     def _get_backend_params(
         self,
-        params: Optional[Union[Dict[str, str], Dict[str, int], Dict[str, float]]],
+        params: Optional[Dict[str, Any]],
     ) -> Dict[str, Any]:
         backend_params = dict(self.params)
         if params is not None:
@@ -71,7 +71,7 @@ def _get_backend_params(
     def _train(
         self,
         corpus: DocumentCorpus,
-        params: Dict[str, Union[int, str]],
+        params: Dict[str, Any],
         jobs: int = 0,
     ) -> None:
         """This method can be overridden by backends. It implements
@@ -81,7 +81,7 @@ def _train(
     def train(
         self,
         corpus: DocumentCorpus,
-        params: Optional[Union[Dict[str, float], Dict[str, int]]] = None,
+        params: Optional[Dict[str, Any]] = None,
         jobs: int = 0,
     ) -> None:
         """Train the model on the given document or subject corpus."""
@@ -116,7 +116,7 @@ def _suggest_batch(
     def suggest(
         self,
         texts: List[str],
-        params: Optional[Union[Dict[str, str], Dict[str, int]]] = None,
+        params: Optional[Dict[str, Any]] = None,
     ) -> SuggestionBatch:
         """Suggest subjects for the input documents and return a list of subject sets
         represented as a list of SubjectSuggestion objects."""
@@ -149,7 +149,7 @@ def _learn(self, corpus, params):
     def learn(
         self,
         corpus: DocumentCorpus,
-        params: Optional[Dict[str, int]] = None,
+        params: Optional[Dict[str, Any]] = None,
     ) -> None:
         """Further train the model on the given document or subject corpus."""
         beparams = self._get_backend_params(params)

diff --git a/annif/backend/dummy.py b/annif/backend/dummy.py
@@ -1,7 +1,7 @@
 """Dummy backend for testing basic interaction of projects and backends"""
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Dict, List, Union
+from typing import TYPE_CHECKING, Any, Dict, List
 
 from annif.suggestion import SubjectSuggestion
 
@@ -24,9 +24,7 @@ def default_params(self) -> Dict[str, int]:
     def initialize(self, parallel: bool = False) -> None:
         self.initialized = True
 
-    def _suggest(
-        self, text: str, params: Dict[str, Union[int, str]]
-    ) -> List[SubjectSuggestion]:
+    def _suggest(self, text: str, params: Dict[str, Any]) -> List[SubjectSuggestion]:
         score = float(params.get("score", 1.0))
 
         # Ensure tests fail if "text" with wrong type ends up here
@@ -47,7 +45,7 @@ def _suggest(
     def _learn(
         self,
         corpus: DocumentCorpus,
-        params: Dict[str, Union[int, str]],
+        params: Dict[str, Any],
     ) -> None:
         # in this dummy backend we "learn" by picking up the subject ID
         # of the first subject of the first document in the learning set

diff --git a/annif/backend/ensemble.py b/annif/backend/ensemble.py
@@ -1,7 +1,7 @@
 """Ensemble backend that combines results from multiple projects"""
 from __future__ import annotations
 
-from typing import TYPE_CHECKING, Dict, List, Optional, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple
 
 import annif.eval
 import annif.parallel
@@ -49,7 +49,7 @@ def _merge_source_batches(
         self,
         batch_by_source: Dict[str, SuggestionBatch],
         sources: List[Tuple[str, float]],
-        params: Dict[str, Union[int, str]],
+        params: Dict[str, Any],
     ) -> SuggestionBatch:
         """Merge the given SuggestionBatches from each source into a single
         SuggestionBatch. The default implementation computes a weighted
@@ -63,7 +63,7 @@ def _merge_source_batches(
         )
 
     def _suggest_batch(
-        self, texts: List[str], params: Dict[str, Union[float, str]]
+        self, texts: List[str], params: Dict[str, Any]
     ) -> SuggestionBatch:
         sources = annif.util.parse_sources(params["sources"])
         batch_by_source = self._suggest_with_sources(texts, sources)
@@ -159,7 +159,5 @@ def get_hp_optimizer(
     ) -> EnsembleOptimizer:
         return EnsembleOptimizer(self, corpus, metric)
 
-    def _train(
-        self, corpus: DocumentCorpus, params: Dict[str, Union[int, str]], jobs: int = 0
-    ):
+    def _train(self, corpus: DocumentCorpus, params: Dict[str, Any], jobs: int = 0):
         raise NotSupportedException("Training ensemble backend is not possible.")
diff --git a/annif/backend/fasttext.py b/annif/backend/fasttext.py
@@ -3,7 +3,7 @@
 
 import collections
 import os.path
-from typing import TYPE_CHECKING, Dict, List, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Tuple
 
 import fasttext
 
@@ -56,7 +56,7 @@ class FastTextBackend(mixins.ChunkingBackend, backend.AnnifBackend):
     # defaults for uninitialized instances
     _model = None
 
-    def default_params(self) -> Dict[str, Union[float, str]]:
+    def default_params(self) -> Dict[str, Any]:
         params = backend.AnnifBackend.DEFAULT_PARAMETERS.copy()
         params.update(mixins.ChunkingBackend.DEFAULT_PARAMETERS)
         params.update(self.DEFAULT_PARAMETERS)
@@ -119,7 +119,7 @@ def _create_train_file(
             corpus, self.datadir, self.TRAIN_FILE, method=self._write_train_file
         )
 
-    def _create_model(self, params: Dict[str, Union[float, str]], jobs: int) -> None:
+    def _create_model(self, params: Dict[str, Any], jobs: int) -> None:
         self.info("creating fastText model")
         trainpath = os.path.join(self.datadir, self.TRAIN_FILE)
         modelpath = os.path.join(self.datadir, self.MODEL_FILE)
@@ -137,7 +137,7 @@ def _create_model(self, params: Dict[str, Union[float, str]], jobs: int) -> None
     def _train(
         self,
         corpus: DocumentCorpus,
-        params: Dict[str, Union[float, str]],
+        params: Dict[str, Any],
         jobs: int = 0,
     ) -> None:
         if corpus != "cached":
@@ -163,7 +163,7 @@ def _predict_chunks(
         )
 
     def _suggest_chunks(
-        self, chunktexts: List[str], params: Dict[str, Union[float, str]]
+        self, chunktexts: List[str], params: Dict[str, Any]
     ) -> List[SubjectSuggestion]:
         limit = int(params["limit"])
         chunklabels, chunkscores = self._predict_chunks(chunktexts, limit)

diff --git a/annif/backend/http.py b/annif/backend/http.py
@@ -3,7 +3,7 @@
 from __future__ import annotations
 
 import importlib
-from typing import TYPE_CHECKING, Dict, List, Optional, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Optional, Union
 
 import dateutil.parser
 import requests
@@ -63,9 +63,7 @@ def _get_project_info(self, key: str) -> Optional[Union[bool, str]]:
         else:
             return None
 
-    def _suggest(
-        self, text: str, params: Dict[str, Union[int, str]]
-    ) -> List[SubjectSuggestion]:
+    def _suggest(self, text: str, params: Dict[str, Any]) -> List[SubjectSuggestion]:
         data = {"text": text}
         if "project" in params:
             data["project"] = params["project"]

diff --git a/annif/backend/mllm.py b/annif/backend/mllm.py
@@ -2,7 +2,7 @@
 from __future__ import annotations
 
 import os.path
-from typing import TYPE_CHECKING, Dict, Iterator, List, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, Iterator, List, Tuple
 
 import joblib
 import numpy as np
@@ -93,7 +93,7 @@ class MLLMBackend(hyperopt.AnnifHyperoptBackend):
     def get_hp_optimizer(self, corpus: DocumentCorpus, metric: str) -> MLLMOptimizer:
         return MLLMOptimizer(self, corpus, metric)
 
-    def default_params(self) -> Dict[str, Union[float, bool]]:
+    def default_params(self) -> Dict[str, Any]:
         params = backend.AnnifBackend.DEFAULT_PARAMETERS.copy()
         params.update(self.DEFAULT_PARAMETERS)
         return params
@@ -124,7 +124,7 @@ def initialize(self, parallel: bool = False) -> None:
     def _train(
         self,
         corpus: DocumentCorpus,
-        params: Dict[str, Union[float, bool, str]],
+        params: Dict[str, Any],
         jobs: int = 0,
     ) -> None:
         self.info("starting train")
@@ -158,16 +158,14 @@ def _generate_candidates(self, text: str) -> List[Candidate]:
     def _prediction_to_result(
         self,
         prediction: List[Tuple[np.float64, int]],
-        params: Dict[str, Union[float, bool, str]],
+        params: Dict[str, Any],
     ) -> Iterator:
         vector = np.zeros(len(self.project.subjects), dtype=np.float32)
         for score, subject_id in prediction:
             vector[subject_id] = score
         return vector_to_suggestions(vector, int(params["limit"]))
 
-    def _suggest(
-        self, text: str, params: Dict[str, Union[float, bool, str]]
-    ) -> Iterator:
+    def _suggest(self, text: str, params: Dict[str, Any]) -> Iterator:
         candidates = self._generate_candidates(text)
         prediction = self._model.predict(candidates)
         return self._prediction_to_result(prediction, params)
diff --git a/annif/backend/nn_ensemble.py b/annif/backend/nn_ensemble.py
@@ -5,7 +5,7 @@
 import os.path
 import shutil
 from io import BytesIO
-from typing import TYPE_CHECKING, Dict, List, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Tuple, Union
 
 import joblib
 import lmdb
@@ -112,7 +112,7 @@ class NNEnsembleBackend(backend.AnnifLearningBackend, ensemble.BaseEnsembleBacke
     # defaults for uninitialized instances
     _model = None
 
-    def default_params(self) -> Dict[str, Union[float, str]]:
+    def default_params(self) -> Dict[str, Any]:
         params = backend.AnnifBackend.DEFAULT_PARAMETERS.copy()
         params.update(self.DEFAULT_PARAMETERS)
         return params
@@ -140,7 +140,7 @@ def _merge_source_batches(
         self,
         batch_by_source: Dict[str, SuggestionBatch],
         sources: List[Tuple[str, float]],
-        params: Dict[str, Union[float, str]],
+        params: Dict[str, Any],
     ) -> SuggestionBatch:
         src_weight = dict(sources)
         score_vectors = np.array(
@@ -199,7 +199,7 @@ def _create_model(self, sources: List[Tuple[str, float]]) -> None:
     def _train(
         self,
         corpus: DocumentCorpus,
-        params: Dict[str, Union[float, str]],
+        params: Dict[str, Any],
         jobs: int = 0,
     ) -> None:
         sources = annif.util.parse_sources(self.params["sources"])
@@ -286,7 +286,7 @@ def _fit_model(
     def _learn(
         self,
         corpus: DocumentCorpus,
-        params: Dict[str, Union[float, str]],
+        params: Dict[str, Any],
     ) -> None:
         self.initialize()
         self._fit_model(

diff --git a/annif/backend/omikuji.py b/annif/backend/omikuji.py
@@ -3,7 +3,7 @@
 
 import os.path
 import shutil
-from typing import TYPE_CHECKING, Dict, List, Union
+from typing import TYPE_CHECKING, Any, Dict, List
 
 import omikuji
 
@@ -43,7 +43,7 @@ class OmikujiBackend(mixins.TfidfVectorizerMixin, backend.AnnifBackend):
         "collapse_every_n_layers": 0,
     }
 
-    def default_params(self) -> Dict[str, Union[int, bool]]:
+    def default_params(self) -> Dict[str, Any]:
         params = backend.AnnifBackend.DEFAULT_PARAMETERS.copy()
         params.update(self.DEFAULT_PARAMETERS)
         return params
@@ -96,7 +96,7 @@ def _create_train_file(self, veccorpus: csr_matrix, corpus: DocumentCorpus) -> N
             trainfile.seek(0)
             print("{:08d}".format(n_samples), end="", file=trainfile)
 
-    def _create_model(self, params: Dict[str, Union[int, bool]], jobs: int) -> None:
+    def _create_model(self, params: Dict[str, Any], jobs: int) -> None:
         train_path = os.path.join(self.datadir, self.TRAIN_FILE)
         model_path = os.path.join(self.datadir, self.MODEL_FILE)
         hyper_param = omikuji.Model.default_hyper_param()
@@ -114,7 +114,7 @@ def _create_model(self, params: Dict[str, Union[int, bool]], jobs: int) -> None:
     def _train(
         self,
         corpus: DocumentCorpus,
-        params: Dict[str, Union[int, bool]],
+        params: Dict[str, Any],
         jobs: int = 0,
     ) -> None:
         if corpus != "cached":
@@ -135,7 +135,7 @@ def _train(
         self._create_model(params, jobs)
 
     def _suggest_batch(
-        self, texts: List[str], params: Dict[str, Union[int, bool]]
+        self, texts: List[str], params: Dict[str, Any]
     ) -> SuggestionBatch:
         vector = self.vectorizer.transform(texts)
         limit = int(params["limit"])

diff --git a/annif/backend/pav.py b/annif/backend/pav.py
@@ -5,7 +5,7 @@
 from __future__ import annotations
 
 import os.path
-from typing import TYPE_CHECKING, Dict, List, Tuple, Union
+from typing import TYPE_CHECKING, Any, Dict, List, Tuple
 
 import joblib
 import numpy as np
@@ -36,7 +36,7 @@ class PAVBackend(ensemble.BaseEnsembleBackend):
 
     DEFAULT_PARAMETERS = {"min-docs": 10}
 
-    def default_params(self) -> Dict[str, int]:
+    def default_params(self) -> Dict[str, Any]:
         params = backend.AnnifBackend.DEFAULT_PARAMETERS.copy()
         params.update(self.DEFAULT_PARAMETERS)
         return params
@@ -67,7 +67,7 @@ def _merge_source_batches(
         self,
         batch_by_source: Dict[str, SuggestionBatch],
         sources: List[Tuple[str, float]],
-        params: Dict[str, Union[int, str]],
+        params: Dict[str, Any],
     ) -> SuggestionBatch:
         reg_batch_by_source = {}
         for project_id, batch in batch_by_source.items():
@@ -156,7 +156,7 @@ def _create_pav_model(
     def _train(
         self,
         corpus: DocumentCorpus,
-        params: Dict[str, Union[int, str]],
+        params: Dict[str, Any],
         jobs: int = 0,
     ) -> None:
         if corpus == "cached":