Make static type checking work (#578)

* Type hints in undodb

* api.resources.emit

* Add future import

* More typing fixes

* More type fixes

* type fixes in image

* More type fixes

* Fix tets

* Add missing import

* More type fixes

* More type fixes

* More type fixes

* One more fix for today.

* More type fixes

* More fixes

* More fixes

* Fix failing test

* Fix types in trees resouce

* Fix types in dbmanager

* Fix regression in trees

* More fixes

* Fixes in metadata

* More type fixes

* More fixes

* More fixes

* 1 more fix

* Fixes in resources.util

* Fixes in search.indexer

* Ignore two errors

* Remove stray imports, fix 1 error

* Add mypy to CI

* Try to fix workflow

* Final type fixes

* Try to fix type

* fix image

Author: DavidMStraub

.github/workflows/test.yml

@@ -25,5 +25,7 @@ jobs:
           pip install -r requirements-dev.txt
           pip install .[ai]
           pip list
+      - name: Run type checker
+        run: mypy --ignore-missing-imports --exclude build .
       - name: Test with pytest
         run: pytest

alembic_users/env.py

@@ -1,3 +1,5 @@
+# type: ignore
+
 import os
 from logging.config import fileConfig
 

gramps_webapi/__main__.py

@@ -31,10 +31,10 @@
 from threading import Thread
 
 import click
-import waitress
+import waitress  # type: ignore
 import webbrowser
 
-from .api.search import get_search_indexer
+from .api.search import get_search_indexer, get_semantic_search_indexer
 from .api.util import get_db_manager, list_trees, close_db
 from .app import create_app
 from .auth import add_user, delete_user, fill_tree, user_db
@@ -213,7 +213,10 @@ def search(ctx, tree, semantic):
         tree = dbmgr.dirname
     with app.app_context():
         ctx.obj["db_manager"] = get_db_manager(tree=tree)
-        ctx.obj["search_indexer"] = get_search_indexer(tree=tree, semantic=semantic)
+        if semantic:
+            ctx.obj["search_indexer"] = get_semantic_search_indexer(tree=tree)
+        else:
+            ctx.obj["search_indexer"] = get_search_indexer(tree=tree)
 
 
 def progress_callback_count(

gramps_webapi/api/export.py

@@ -21,6 +21,8 @@
 
 """Functions for exporting a database."""
 
+from __future__ import annotations
+
 import mimetypes
 import os
 import uuid
@@ -42,6 +44,7 @@
 )
 from gramps.gen.user import User
 from gramps.gen.utils.resourcepath import ResourcePath
+from torch import Value
 
 from ..const import DISABLED_EXPORTERS
 from .util import UserTaskProgress, abort_with_message, get_locale_for_language
@@ -188,7 +191,7 @@ def apply_proxy(self, proxy_name, dbase):
         return dbase
 
 
-def get_exporters(extension: str = None):
+def get_exporters(extension: str | None = None):
     """Extract and return list of exporters."""
     exporters = []
     plugin_manager = BasePluginManager.get_instance()
@@ -267,6 +270,8 @@ def run_export(
 ):
     """Generate the export."""
     export_path = current_app.config.get("EXPORT_DIR")
+    if not export_path:
+        raise abort_with_message(500, "EXPORT_DIR not set in configuration")
     os.makedirs(export_path, exist_ok=True)
     file_name = f"{uuid.uuid4()}.{extension}"
     file_path = os.path.join(export_path, file_name)

gramps_webapi/api/file.py

@@ -52,12 +52,13 @@ def __init__(self, handle, db_handle: DbReadBase):
         self.path = self.media.path
         self.checksum = self.media.checksum
 
-    def _get_media_object(self) -> Optional[Media]:
+    def _get_media_object(self) -> Media:
         """Get the media object from the database."""
         try:
             return self.db_handle.get_media_from_handle(self.handle)
         except HandleError:
-            return None
+            abort_with_message(404, "Media object not found")
+            raise  # unreachable - for type checker
 
     def get_file_object(self) -> BinaryIO:
         """Return a binary file object."""

gramps_webapi/api/html.py

@@ -23,8 +23,8 @@
 
 from typing import Callable, Optional
 
-import bleach
-from bleach.css_sanitizer import CSSSanitizer
+import bleach  # type: ignore
+from bleach.css_sanitizer import CSSSanitizer  # type: ignore
 from gramps.gen.errors import HandleError
 from gramps.gen.lib import Note, NoteType, StyledText
 from gramps.plugins.lib.libhtml import Html

gramps_webapi/api/image.py

@@ -29,6 +29,7 @@
 import ffmpeg
 from pdf2image import convert_from_path
 from PIL import Image, ImageOps
+from PIL.Image import Image as ImageType
 from pkg_resources import resource_filename
 
 from gramps_webapi.const import MIME_PDF
@@ -37,12 +38,13 @@
 from .util import abort_with_message
 
 
-def image_thumbnail(image: Image, size: int, square: bool = False) -> Image:
+def image_thumbnail(image: ImageType, size: int, square: bool = False) -> ImageType:
     """Return a thumbnail of `size` (longest side) for the image.
 
     If `square` is true, the image is cropped to a centered square.
     """
     img = ImageOps.exif_transpose(image)
+    assert img is not None, "img is None"  # for type checker. Can't happen in practice.
     if square:
         # don't enlarge image: square size is at most shorter (!) side's length
         size_orig = min(img.size)
@@ -52,25 +54,25 @@ def image_thumbnail(image: Image, size: int, square: bool = False) -> Image:
             (size_square, size_square),
             bleed=0.0,
             centering=(0.5, 0.5),
-            method=Image.BICUBIC,
+            method=Image.Resampling.BICUBIC,
         )
     img.thumbnail((size, size))
     return img
 
 
-def image_square(image: Image) -> Image:
+def image_square(image: ImageType) -> ImageType:
     """Crop an image to a centered square."""
     size = min(image.size)
     return ImageOps.fit(
         image,
         (size, size),
         bleed=0.0,
         centering=(0.0, 0.5),
-        method=Image.BICUBIC,
+        method=Image.Resampling.BICUBIC,
     )
 
 
-def crop_image(image: Image, x1: int, y1: int, x2: int, y2: int) -> Image:
+def crop_image(image: ImageType, x1: int, y1: int, x2: int, y2: int) -> ImageType:
     """Crop an image.
 
     The arguments `x1`, `y1`, `x2`, `y2` are the coordinates of the cropped region
@@ -84,7 +86,7 @@ def crop_image(image: Image, x1: int, y1: int, x2: int, y2: int) -> Image:
     return image.crop((x1_abs, y1_abs, x2_abs, y2_abs))
 
 
-def save_image_buffer(image: Image, fmt="JPEG") -> BinaryIO:
+def save_image_buffer(image: ImageType, fmt="JPEG") -> BinaryIO:
     """Save an image to a binary buffer."""
     buffer = io.BytesIO()
     if image.mode != "RGB":
@@ -118,7 +120,7 @@ def __init__(self, stream: BinaryIO, mime_type: str) -> None:
             self.is_image = False
             self.is_video = False
 
-    def get_image(self) -> Image:
+    def get_image(self) -> ImageType:
         """Get a Pillow Image instance."""
         if self.mime_type == MIME_PDF:
             return self._get_image_pdf()
@@ -142,7 +144,7 @@ def get_cropped(
             img = image_square(img)
         return save_image_buffer(img)
 
-    def _get_image_pdf(self) -> Image:
+    def _get_image_pdf(self) -> ImageType:
         """Get a Pillow Image instance of the PDF's first page."""
         ims = self._apply_to_path(
             convert_from_path, single_file=True, use_cropbox=True, dpi=100
@@ -165,7 +167,7 @@ def _apply_to_path(self, func: Callable, *args, **kwargs):
             os.remove(temp_filename)
         return output
 
-    def _get_image_video(self) -> Image:
+    def _get_image_video(self) -> ImageType:
         """Get a Pillow Image instance of the video's first frame."""
         out, _ = self._apply_to_path(
             lambda path: (
@@ -248,7 +250,7 @@ def detect_faces(stream: BinaryIO) -> list[tuple[float, float, float, float]]:
     # Extract and normalize face bounding boxes
     detected_faces = []
     for face in faces[1]:
-        x, y, w, h = face[:4]
+        x, y, w, h = np.asarray(face)[:4]
         x = float(x)
         y = float(y)
         w = float(w)

gramps_webapi/api/llm/__init__.py

@@ -5,7 +5,7 @@
 from flask import current_app
 from openai import OpenAI, RateLimitError, APIError
 
-from ..search import get_search_indexer
+from ..search import get_semantic_search_indexer
 from ..util import abort_with_message, get_logger
 
 
@@ -50,10 +50,11 @@ def answer_prompt(prompt: str, system_prompt: str, config: dict | None = None) -
 
     client = get_client(config=config)
     model = config.get("LLM_MODEL")
+    assert model is not None, "No LLM model specified"  # mypy; shouldn't happen
 
     try:
         response = client.chat.completions.create(
-            messages=messages,
+            messages=messages,  # type: ignore
             model=model,
         )
     except RateLimitError:
@@ -64,9 +65,10 @@ def answer_prompt(prompt: str, system_prompt: str, config: dict | None = None) -
         abort_with_message(500, "Unexpected error.")
 
     try:
-        answer = response.to_dict()["choices"][0]["message"]["content"]
+        answer = response.to_dict()["choices"][0]["message"]["content"]  # type: ignore
     except (KeyError, IndexError):
         abort_with_message(500, "Error parsing chat API response.")
+        raise  # mypy; unreachable
 
     return sanitize_answer(answer)
 
@@ -102,7 +104,7 @@ def contextualize_prompt(prompt: str, context: str) -> str:
 
 
 def retrieve(tree: str, prompt: str, include_private: bool, num_results: int = 10):
-    searcher = get_search_indexer(tree, semantic=True)
+    searcher = get_semantic_search_indexer(tree)
     total, hits = searcher.search(
         query=prompt,
         page=1,

gramps_webapi/api/media.py

@@ -41,6 +41,7 @@
     get_db_handle,
     get_db_outside_request,
     get_tree_from_jwt,
+    get_tree_from_jwt_or_fail,
 )
 
 PREFIX_S3 = "s3://"
@@ -351,9 +352,10 @@ def update_usage_media(
 ) -> int:
     """Update the usage of media."""
     if not tree:
-        tree = get_tree_from_jwt()
+        tree = get_tree_from_jwt_or_fail()
     if not user_id:
         user_id = get_jwt_identity()
+    assert user_id is not None, "Unexpected error while looking up user ID."
     db_handle = get_db_outside_request(
         tree=tree, view_private=True, readonly=True, user_id=user_id
     )
@@ -371,11 +373,12 @@ def check_quota_media(
 ) -> None:
     """Check whether the quota allows adding `to_add` bytes and abort if not."""
     if not tree:
-        tree = get_tree_from_jwt()
+        tree = get_tree_from_jwt_or_fail()
     usage_dict = get_tree_usage(tree)
     if not usage_dict or usage_dict.get("usage_media") is None:
         update_usage_media(tree=tree, user_id=user_id)
-    usage_dict = get_tree_usage(tree)
+        usage_dict = get_tree_usage(tree)
+        assert usage_dict is not None, "Unexpected error while looking up usage data."
     usage = usage_dict["usage_media"]
     quota = usage_dict.get("quota_media")
     if quota is None:

gramps_webapi/api/media_importer.py

@@ -77,7 +77,7 @@ def _identify_missing_files(self) -> MissingFiles:
             obj for obj in self.objects if obj.handle not in handles_existing
         ]
 
-        missing_files = {}
+        missing_files: dict[str, list[dict[str, str]]] = {}
         for obj in objects_missing:
             if obj.checksum not in missing_files:
                 missing_files[obj.checksum] = []