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