Dateien nach "tests" hochladen

tests/ensure_indexes_and_show.py

@@ -1,12 +1,21 @@
 #!/usr/bin/env python3
 # -*- coding: utf-8 -*-
+"""
+tests/ensure_indexes_and_show.py (compat v1.0.1)
+
+Erzwingt ensure_payload_indexes(...) und zeigt danach payload_schema je Collection.
+Kompatibel mit qdrant-client Versionen ohne 'with_payload_schema' im Wrapper.
+"""
 from __future__ import annotations
+
 import json
+import os
+import urllib.request
+import urllib.error
+from typing import Any, Dict
+
 from app.core.qdrant import QdrantConfig, get_client, ensure_payload_indexes, collection_names
 
-from __future__ import annotations
-import json, os, urllib.request, urllib.error
-from typing import Any, Dict
 
 def _safe_model_dump(obj: Any) -> Dict[str, Any]:
     if hasattr(obj, "model_dump"):
@@ -15,6 +24,7 @@ def _safe_model_dump(obj: Any) -> Dict[str, Any]:
         return obj.dict()
     return obj if isinstance(obj, dict) else {}
 
+
 def _cfg_base_url(cfg) -> str:
     if getattr(cfg, "url", None):
         return cfg.url.rstrip("/")
@@ -22,6 +32,7 @@ def _cfg_base_url(cfg) -> str:
     port = getattr(cfg, "port", None) or os.getenv("QDRANT_PORT") or "6333"
     return f"http://{host}:{port}"
 
+
 def _http_get(url: str, api_key: str | None) -> Dict[str, Any]:
     req = urllib.request.Request(url, method="GET")
     if api_key:
@@ -32,7 +43,9 @@ def _http_get(url: str, api_key: str | None) -> Dict[str, Any]:
         data = resp.read()
         return json.loads(data.decode("utf-8"))
 
+
 def get_collection_payload_schema(client, cfg, name: str) -> Dict[str, Any] | None:
+    # Strategy 1: OpenAPI client with flag
     try:
         oc = getattr(client, "openapi_client", None)
         if oc is not None and hasattr(oc, "collections_api"):
@@ -42,6 +55,7 @@ def get_collection_payload_schema(client, cfg, name: str) -> Dict[str, Any] | No
             return (d.get("result") or {}).get("payload_schema")
     except Exception:
         pass
+    # Strategy 2: wrapper
     try:
         info = client.get_collection(collection_name=name)
         d = _safe_model_dump(info)
@@ -50,6 +64,7 @@ def get_collection_payload_schema(client, cfg, name: str) -> Dict[str, Any] | No
             return ps
     except Exception:
         pass
+    # Strategy 3: direct HTTP
     try:
         base = _cfg_base_url(cfg)
         url = f"{base}/collections/{name}?with_payload_schema=true"
@@ -58,6 +73,7 @@ def get_collection_payload_schema(client, cfg, name: str) -> Dict[str, Any] | No
     except Exception:
         return None
 
+
 def main():
     cfg = QdrantConfig.from_env()
     client = get_client(cfg)
@@ -68,5 +84,6 @@ def main():
         res[name] = get_collection_payload_schema(client, cfg, name)
     print(json.dumps(res, ensure_ascii=False, indent=2))
 
+
 if __name__ == "__main__":
     main()