examples: review fixes\n- image_search/main: use CLIP_MODEL_DIMENSION in Literal via type alias for mypy; preserve schema inference\n- image_search/colpali_main: decorator with targeted untyped-decorator ignore\n- pdf_embedding/main: targeted type ignore on return instead of cast

Author: samojavo

examples/image_search/colpali_main.py

@@ -100,6 +100,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 
 
 # --- Search API ---
+@app.get("/search")  # type: ignore
 def search(
     q: str = Query(..., description="Search query"),
     limit: int = Query(5, description="Number of results"),
@@ -133,5 +134,3 @@ def search(
     }
 
 
-# Attach route without using decorator to avoid untyped-decorator when FastAPI types are unavailable
-app.get("/search")(search)

examples/image_search/main.py

@@ -3,7 +3,7 @@
 import io
 import os
 from contextlib import asynccontextmanager
-from typing import Any, cast, AsyncIterator
+from typing import Any, cast, AsyncIterator, Literal, Final, TYPE_CHECKING
 
 import cocoindex
 import torch
@@ -19,7 +19,11 @@
 QDRANT_URL = os.getenv("QDRANT_URL", "http://localhost:6334/")
 QDRANT_COLLECTION = "ImageSearch"
 CLIP_MODEL_NAME = "openai/clip-vit-large-patch14"
-# Using simple list[float] for embeddings for readability in example code.
+CLIP_MODEL_DIMENSION: Final[int] = 768
+if TYPE_CHECKING:
+    ClipDim = Literal[768]
+else:
+    ClipDim = int
 
 
 @functools.cache
@@ -29,21 +33,24 @@ def get_clip_model() -> tuple[CLIPModel, CLIPProcessor]:
     return model, processor
 
 
-def embed_query(text: str) -> list[float]:
+def embed_query(text: str) -> cocoindex.Vector[cocoindex.Float32, ClipDim]:
     """
     Embed the caption using CLIP model.
     """
     model, processor = get_clip_model()
     inputs = processor(text=[text], return_tensors="pt", padding=True)
     with torch.no_grad():
         features = model.get_text_features(**inputs)
-    return cast(list[float], features[0].tolist())
+    return cast(
+        cocoindex.Vector[cocoindex.Float32, ClipDim],
+        features[0].tolist(),
+    )
 
 
 @cocoindex.op.function(cache=True, behavior_version=1, gpu=True)
 def embed_image(
     img_bytes: bytes,
-) -> list[float]:
+) -> cocoindex.Vector[cocoindex.Float32, ClipDim]:
     """
     Convert image to embedding using CLIP model.
     """
@@ -52,7 +59,10 @@ def embed_image(
     inputs = processor(images=image, return_tensors="pt")
     with torch.no_grad():
         features = model.get_image_features(**inputs)
-    return cast(list[float], features[0].tolist())
+    return cast(
+        cocoindex.Vector[cocoindex.Float32, ClipDim],
+        features[0].tolist(),
+    )
 
 
 # CocoIndex flow: Ingest images, extract captions, embed, export to Qdrant
@@ -141,7 +151,7 @@ async def lifespan(app: FastAPI) -> AsyncIterator[None]:
 
 
 # --- Search API ---
-@app.get("/search")  # type: ignore
+@app.get("/search")  # type: ignore[untyped-decorator]
 def search(
     q: str = Query(..., description="Search query"),
     limit: int = Query(5, description="Number of results"),
@@ -170,5 +180,3 @@ def search(
         ]
     }
 
-
-# Route attached via decorator above for readability

examples/pdf_embedding/main.py

@@ -9,7 +9,7 @@
 from marker.output import text_from_rendered
 from psycopg_pool import ConnectionPool
 from jinja2 import Template
-from typing import Any, cast
+from typing import Any
 
 
 class PdfToMarkdown(cocoindex.op.FunctionSpec):
@@ -34,8 +34,7 @@ def __call__(self, content: bytes) -> str:
             temp_file.write(content)
             temp_file.flush()
             text_any, _, _ = text_from_rendered(self._converter(temp_file.name))
-            text: str = cast(str, text_any)
-            return text
+            return text_any  # type: ignore
 
 
 @cocoindex.transform_flow()