diff --git a/tests/ensure_indexes_and_show.py b/tests/ensure_indexes_and_show.py index a8183d5..01ba553 100644 --- a/tests/ensure_indexes_and_show.py +++ b/tests/ensure_indexes_and_show.py @@ -1,14 +1,63 @@ #!/usr/bin/env python3 # -*- coding: utf-8 -*- -""" -tests/ensure_indexes_and_show.py (v1.0) -Alias für die Test-Suite: ruft ensure_payload_indexes(...) auf -und zeigt danach das payload_schema (mit with_payload_schema=True). -""" from __future__ import annotations import json 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"): + return obj.model_dump() + if hasattr(obj, "dict"): + 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("/") + host = getattr(cfg, "host", None) or os.getenv("QDRANT_HOST") or "127.0.0.1" + 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: + req.add_header("api-key", api_key) + req.add_header("Authorization", f"Bearer {api_key}") + req.add_header("Accept", "application/json") + with urllib.request.urlopen(req, timeout=30) as resp: + data = resp.read() + return json.loads(data.decode("utf-8")) + +def get_collection_payload_schema(client, cfg, name: str) -> Dict[str, Any] | None: + try: + oc = getattr(client, "openapi_client", None) + if oc is not None and hasattr(oc, "collections_api"): + api = oc.collections_api + info = api.get_collection(collection_name=name, with_payload_schema=True) + d = _safe_model_dump(info) + return (d.get("result") or {}).get("payload_schema") + except Exception: + pass + try: + info = client.get_collection(collection_name=name) + d = _safe_model_dump(info) + ps = (d.get("result") or {}).get("payload_schema") + if ps is not None: + return ps + except Exception: + pass + try: + base = _cfg_base_url(cfg) + url = f"{base}/collections/{name}?with_payload_schema=true" + raw = _http_get(url, getattr(cfg, "api_key", None)) + return (raw.get("result") or {}).get("payload_schema") + except Exception: + return None + def main(): cfg = QdrantConfig.from_env() client = get_client(cfg) @@ -16,9 +65,7 @@ def main(): notes, chunks, edges = collection_names(cfg.prefix) res = {} for name in (notes, chunks, edges): - info = client.get_collection(collection_name=name, with_payload_schema=True) - d = info.model_dump() if hasattr(info, "model_dump") else (info.dict() if hasattr(info, "dict") else info) - res[name] = (d.get("result") or {}).get("payload_schema") + res[name] = get_collection_payload_schema(client, cfg, name) print(json.dumps(res, ensure_ascii=False, indent=2)) if __name__ == "__main__":