feat(config): add typed resource schema

2026-07-11 20:43:08 +02:00
parent fe8d70da47
commit 1e0911bb6a
5 changed files with 380 additions and 45 deletions
@@ -0,0 +1,74 @@
+from __future__ import annotations
+
+from copy import deepcopy
+from typing import Any
+
+
+_LEGACY_RESOURCE_KEYS = {
+    "database",
+    "rest",
+    "vector_db",
+    "vector_rest",
+    "vector_write_rest",
+    "paths",
+}
+
+
+def _as_mapping(value: Any) -> dict[str, Any] | None:
+    if isinstance(value, dict):
+        return deepcopy(value)
+    model_dump = getattr(value, "model_dump", None)
+    if callable(model_dump):
+        return model_dump(by_alias=True)
+    return None
+
+
+def translate_legacy_config(raw: dict[str, Any]) -> tuple[dict[str, Any], bool]:
+    """Translate the legacy flat resource keys without validating their contents."""
+    translated = deepcopy(raw)
+    legacy = any(key in raw for key in _LEGACY_RESOURCE_KEYS)
+    if not legacy:
+        return translated, False
+
+    database = _as_mapping(raw.get("database"))
+    rest = raw.get("rest")
+    if "dwh" not in translated and database is not None:
+        if database.get("transport", "direct") == "rest":
+            identity = {
+                key: database[key]
+                for key in ("database", "schema")
+                if key in database
+            }
+            translated["dwh"] = {
+                "type": "thoth_rest",
+                "database": identity,
+                "endpoint": rest,
+            }
+        else:
+            connection = database
+            connection.pop("transport", None)
+            translated["dwh"] = {
+                "type": "postgres_direct",
+                "connection": connection,
+            }
+
+    if "vectors" not in translated:
+        vector_db = raw.get("vector_db")
+        reader = raw.get("vector_rest")
+        writer = raw.get("vector_write_rest")
+        if reader is not None or writer is not None:
+            translated["vectors"] = {
+                "type": "thoth_vector_http",
+                "reader": reader,
+                "writer": writer,
+                "direct": vector_db,
+            }
+        elif vector_db is not None:
+            translated["vectors"] = {
+                "type": "pgvector_direct",
+                "connection": vector_db,
+            }
+
+    if "roots" not in translated and "paths" in raw:
+        translated["roots"] = deepcopy(raw["paths"])
+    return translated, True