82 lines
2.7 KiB
Python
82 lines
2.7 KiB
Python
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",
|
|
"resources",
|
|
}
|
|
|
|
|
|
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)
|
|
resources = _as_mapping(raw.get("resources"))
|
|
if isinstance(resources, dict) and "embeddings" in resources and "embeddings" not in translated:
|
|
embedding = _as_mapping(resources.get("embeddings"))
|
|
if isinstance(embedding, dict):
|
|
translated["embeddings"] = embedding
|
|
if "dimensions" in translated["embeddings"] and "dim" not in translated["embeddings"]:
|
|
translated["embeddings"]["dim"] = translated["embeddings"].pop("dimensions")
|
|
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
|