perf(codes): cache terminologies (#108)

* refactor(cli): only load one matcher in search-codes

* perf(codes): cache code matchers by source file

* fix(codes): create cache dir if missing

* chore(docker): pin python to 3.12 in runtime layer

Author: cmdoret

Dockerfile

@@ -1,17 +1,15 @@
 ARG VERSION_BUILD
 
-FROM python:3.11-slim
 
-ENV PYTHONUNBUFFERED=true
-WORKDIR /app
 
 ##################################################
 # Poetry setup
 ##################################################
-FROM python AS poetry
+FROM python:3.12-slim AS poetry
 
 WORKDIR /app
 # Install poetry
+ENV PYTHONUNBUFFERED=true
 ENV POETRY_HOME=/opt/poetry
 ENV POETRY_VIRTUALENVS_IN_PROJECT=true
 ENV PATH="$POETRY_HOME/bin:$PATH"
@@ -32,7 +30,7 @@ RUN poetry install --no-interaction --no-ansi -vvv
 ##################################################
 # modos setup
 ##################################################
-FROM python AS runtime
+FROM python:3.12-slim AS runtime
 ARG VERSION_BUILD
 ENV PATH="/app/.venv/bin:$PATH"
 COPY --from=poetry /app /app

modos/cli.py

@@ -24,7 +24,7 @@
 from modos.genomics.htsget import HtsgetConnection
 from modos.genomics.region import Region
 from modos.io import parse_instance
-from modos.prompt import SlotPrompter
+from modos.prompt import SlotPrompter, fuzzy_complete
 from modos.remote import EndpointManager, list_remote_items
 from modos.storage import connect_s3
 
@@ -308,17 +308,19 @@ def search_codes(
     ] = 50,
 ):
     """Search for terminology codes using free text."""
-    prompter = SlotPrompter(
-        EndpointManager(ctx.obj.endpoint),
-        prompt=f'Browsing terms for slot "{slot}". Use tab to cycle suggestions.\n> ',
+    matcher = get_slot_matcher(
+        slot,
+        EndpointManager(ctx.obj.endpoint).fuzon,
     )
-    for matcher in prompter.slot_matchers.values():
-        matcher.top = top
+    matcher.top = top
     if query:
-        matches = prompter.slot_matchers[slot].find_codes(query)
+        matches = matcher.find_codes(query)
         out = "\n".join([f"{m.uri} | {m.label}" for m in matches])
     else:
-        out = prompter.prompt_for_slot(slot)
+        out = fuzzy_complete(
+            prompt_txt=f'Browsing terms for slot "{slot}". Use tab to cycle suggestions.\n> ',
+            matcher=matcher,
+        )
     print(out)
 
 

modos/codes.py

@@ -1,13 +1,17 @@
 """Utilities to automatically find / recommend terminology codes from text."""
 from dataclasses import dataclass
-import requests
 from typing import Optional, Protocol
 
+from pathlib import Path
+import requests
+
 
 SLOT_TERMINOLOGIES = {
-    "cell_type": "https://purl.obolibrary.org/obo/cl.owl",
-    "source_material": "https://purl.obolibrary.org/obo/uberon.owl",
-    "taxon_id": "https://purl.obolibrary.org/obo/ncbitaxon/subsets/taxslim.owl",
+    "cell_type": ["https://purl.obolibrary.org/obo/cl.owl"],
+    "source_material": ["https://purl.obolibrary.org/obo/uberon.owl"],
+    "taxon_id": [
+        "https://purl.obolibrary.org/obo/ncbitaxon/subsets/taxslim.owl"
+    ],
 }
 
 
@@ -35,14 +39,21 @@ def __init__(self, slot: str, top: int = 50):
         self.top = top
 
         try:
-            from pyfuzon import TermMatcher
-
-            self.matcher = TermMatcher.from_files([SLOT_TERMINOLOGIES[slot]])
+            from pyfuzon import cache
         except ImportError:
-            raise ValueError(
-                """No endpoint provided and pyfuzon not installed,
-                cannot do code matching."""
+            raise ModuleNotFoundError(
+                "pyfuzon must be installed to perform local code matching."
+            )
+
+        sources = SLOT_TERMINOLOGIES[slot]
+        try:
+            self.matcher = cache.load_by_source(sources)
+        except RuntimeError:
+            Path(cache.get_cache_path(sources)).parent.mkdir(
+                parents=True, exist_ok=True
             )
+            cache.cache_by_source(sources)
+            self.matcher = cache.load_by_source(sources)
 
     def find_codes(self, query: str) -> list[Code]:
         return self.matcher.top(query, self.top)