262 lines
11 KiB
Python
262 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any, Dict, List, Optional
|
|
|
|
from kamidso.contracts.errors import KamidsoError, KamidsoErrorCode
|
|
from kamidso.contracts.user import CurrentUser
|
|
from kamidso.repositories.store import RepositoryStore
|
|
from kamidso.resources.datasource_types import (
|
|
catalog,
|
|
connector_engine_type,
|
|
connector_params,
|
|
normalize_params,
|
|
public_config,
|
|
)
|
|
from kamidso.security.secrets import SecretStore
|
|
|
|
logger = logging.getLogger("kamidso.datasources")
|
|
|
|
|
|
class DatasourceService:
|
|
def __init__(self, store: RepositoryStore, secrets: SecretStore) -> None:
|
|
self._store = store
|
|
self._secrets = secrets
|
|
|
|
def types(self) -> List[Dict[str, Any]]:
|
|
return catalog()
|
|
|
|
async def create(
|
|
self,
|
|
user: CurrentUser,
|
|
*,
|
|
name: str,
|
|
ds_type: str,
|
|
config: Dict[str, Any],
|
|
password: Optional[str] = None,
|
|
test: bool = True,
|
|
) -> Dict[str, Any]:
|
|
label = (name or "").strip()
|
|
if not label:
|
|
raise KamidsoError(KamidsoErrorCode.VALIDATION_ERROR, details="Name is required")
|
|
params = normalize_params(ds_type, config, password)
|
|
if test:
|
|
self.test_params(params)
|
|
credential_ref = None
|
|
if params.get("password"):
|
|
credential_ref = await self._secrets.store(str(params["password"]))
|
|
doc = await self._store.create_datasource(
|
|
self._document(user, label, params, credential_ref)
|
|
)
|
|
return doc
|
|
|
|
async def update(
|
|
self,
|
|
user: CurrentUser,
|
|
datasource_id: str,
|
|
*,
|
|
name: Optional[str] = None,
|
|
ds_type: Optional[str] = None,
|
|
config: Optional[Dict[str, Any]] = None,
|
|
password: Optional[str] = None,
|
|
test: bool = True,
|
|
) -> Dict[str, Any]:
|
|
existing = await self.get(user, datasource_id)
|
|
merged_config = dict(existing.get("config") or {})
|
|
if config:
|
|
merged_config.update(config)
|
|
keep_password = None if password else existing.get("db_pwd") or merged_config.get("password")
|
|
params = normalize_params(
|
|
ds_type or existing.get("type") or existing.get("db_type"),
|
|
merged_config,
|
|
password if password is not None else keep_password,
|
|
)
|
|
if test:
|
|
self.test_params(params)
|
|
credential_ref = existing.get("credential_ref")
|
|
if password:
|
|
credential_ref = await self._secrets.store(password)
|
|
label = (name or "").strip() or existing.get("name") or params["database"]
|
|
payload = self._document(user, label, params, credential_ref)
|
|
payload.pop("workspace_id", None)
|
|
payload.pop("created_by", None)
|
|
updated = await self._store.update_datasource(user.workspace_id, datasource_id, payload)
|
|
if updated is None:
|
|
raise KamidsoError(KamidsoErrorCode.RESOURCE_NOT_FOUND)
|
|
self._invalidate(existing, updated)
|
|
return updated
|
|
|
|
async def delete(self, user: CurrentUser, datasource_id: str) -> None:
|
|
existing = await self.get(user, datasource_id)
|
|
deleted = await self._store.delete_datasource(user.workspace_id, datasource_id)
|
|
if not deleted:
|
|
raise KamidsoError(KamidsoErrorCode.RESOURCE_NOT_FOUND)
|
|
self._invalidate(existing, existing)
|
|
|
|
async def list(self, user: CurrentUser) -> List[Dict[str, Any]]:
|
|
return await self._store.list_datasources(user.workspace_id)
|
|
|
|
async def get(self, user: CurrentUser, datasource_id: str) -> Dict[str, Any]:
|
|
doc = await self._store.get_datasource(user.workspace_id, datasource_id)
|
|
if doc is None:
|
|
raise KamidsoError(KamidsoErrorCode.RESOURCE_NOT_FOUND)
|
|
return doc
|
|
|
|
def test_params(self, params: Dict[str, Any]) -> None:
|
|
try:
|
|
self._connect(params)
|
|
except KamidsoError:
|
|
raise
|
|
except Exception as exc:
|
|
logger.warning("datasource test failed: %s", exc)
|
|
raise KamidsoError(
|
|
KamidsoErrorCode.DATASOURCE_CONNECTION_FAILED,
|
|
str(exc),
|
|
) from exc
|
|
|
|
def test_request(self, ds_type: str, config: Dict[str, Any], password: Optional[str] = None) -> None:
|
|
params = normalize_params(ds_type, config, password)
|
|
self.test_params(params)
|
|
|
|
@staticmethod
|
|
def to_public(doc: Dict[str, Any]) -> Dict[str, Any]:
|
|
config = dict(doc.get("config") or {})
|
|
config.pop("password", None)
|
|
return {
|
|
"id": doc.get("_id"),
|
|
"name": doc.get("name"),
|
|
"type": doc.get("type") or doc.get("db_type"),
|
|
"status": doc.get("status") or "active",
|
|
"access_mode": doc.get("access_mode") or "read_only",
|
|
"description": doc.get("description") or doc.get("comment") or "",
|
|
"database": doc.get("db_name") or config.get("database"),
|
|
"host": doc.get("db_host") or config.get("host"),
|
|
"config": public_config(
|
|
{
|
|
**config,
|
|
"database": doc.get("db_name") or config.get("database"),
|
|
"host": doc.get("db_host") or config.get("host"),
|
|
"port": doc.get("db_port") or config.get("port"),
|
|
"path": doc.get("db_path") or config.get("path"),
|
|
}
|
|
),
|
|
}
|
|
|
|
def _document(
|
|
self,
|
|
user: CurrentUser,
|
|
name: str,
|
|
params: Dict[str, Any],
|
|
credential_ref: Optional[str],
|
|
) -> Dict[str, Any]:
|
|
ds_type = params["type"]
|
|
ext: Dict[str, Any] = {}
|
|
for key in ("schema", "service_name", "sid", "auth_source", "endpoint", "project", "access_id"):
|
|
if params.get(key):
|
|
ext[key] = params[key]
|
|
config = public_config(params)
|
|
if params.get("path"):
|
|
config["path"] = params["path"]
|
|
db_name = params.get("database") or params.get("project") or ""
|
|
if ds_type in {"sqlite", "duckdb", "spark"} and params.get("path"):
|
|
stem = str(params["path"]).replace("\\", "/").rstrip("/").split("/")[-1]
|
|
db_name = f"{ds_type}_{stem.split('.')[0] or 'db'}"
|
|
return {
|
|
"workspace_id": user.workspace_id,
|
|
"name": name,
|
|
"type": ds_type,
|
|
"db_type": ds_type,
|
|
"db_name": db_name,
|
|
"db_host": params.get("host") or params.get("endpoint") or "",
|
|
"db_port": params.get("port") or 0,
|
|
"db_user": params.get("user") or params.get("access_id") or "",
|
|
"db_pwd": params.get("password") or params.get("access_key") or "",
|
|
"db_path": params.get("path") or "",
|
|
"comment": name,
|
|
"description": name,
|
|
"ext_config": ext,
|
|
"config": config,
|
|
"credential_ref": credential_ref,
|
|
"access_mode": "read_only",
|
|
"created_by": user.id,
|
|
"user_id": user.id,
|
|
"status": "active",
|
|
}
|
|
|
|
def _connect(self, params: Dict[str, Any]) -> None:
|
|
ds_type = connector_engine_type(params["type"])
|
|
request_params = connector_params(params)
|
|
try:
|
|
from kamidso.engine.serve.datasource.api.schemas import DatasourceCreateRequest
|
|
from kamidso.engine.serve.datasource.manages.connector_manager import ConnectorManager
|
|
from kamidso.engine.core._private.config import Config
|
|
except Exception as exc: # pragma: no cover - runtime without kamidso.engine.core
|
|
raise KamidsoError(
|
|
KamidsoErrorCode.ENGINE_UNAVAILABLE,
|
|
details=str(exc),
|
|
) from exc
|
|
|
|
cfg = Config()
|
|
system_app = getattr(cfg, "SYSTEM_APP", None)
|
|
if system_app is not None:
|
|
manager = ConnectorManager.get_instance(system_app)
|
|
manager.test_connection(
|
|
DatasourceCreateRequest(type=ds_type, params=request_params)
|
|
)
|
|
return
|
|
self._connect_direct(ds_type, request_params)
|
|
|
|
def _connect_direct(self, ds_type: str, request_params: Dict[str, Any]) -> None:
|
|
import importlib
|
|
|
|
mapping = {
|
|
"mysql": ("kamidso.engine.ext.datasource.rdbms.conn_mysql", "MySQLParameters"),
|
|
"postgresql": ("kamidso.engine.ext.datasource.rdbms.conn_postgresql", "PostgreSQLParameters"),
|
|
"oracle": ("kamidso.engine.ext.datasource.rdbms.conn_oracle", "OracleParameters"),
|
|
"mssql": ("kamidso.engine.ext.datasource.rdbms.conn_mssql", "MSSQLParameters"),
|
|
"sqlite": ("kamidso.engine.ext.datasource.rdbms.conn_sqlite", "SQLiteConnectorParameters"),
|
|
"duckdb": ("kamidso.engine.ext.datasource.rdbms.conn_duckdb", "DuckDbConnectorParameters"),
|
|
"oceanbase": ("kamidso.engine.ext.datasource.rdbms.conn_oceanbase", "OceanBaseParameters"),
|
|
"clickhouse": ("kamidso.engine.ext.datasource.rdbms.conn_clickhouse", "ClickhouseParameters"),
|
|
"starrocks": ("kamidso.engine.ext.datasource.rdbms.conn_starrocks", "StarRocksParameters"),
|
|
"doris": ("kamidso.engine.ext.datasource.rdbms.conn_doris", "DorisParameters"),
|
|
"hive": ("kamidso.engine.ext.datasource.rdbms.conn_hive", "HiveParameters"),
|
|
"gaussdb": ("kamidso.engine.ext.datasource.rdbms.conn_gaussdb", "GaussDBParameters"),
|
|
"openGauss": ("kamidso.engine.ext.datasource.rdbms.conn_openGauss", "openGaussParameters"),
|
|
"vertica": ("kamidso.engine.ext.datasource.rdbms.conn_vertica", "VerticaParameters"),
|
|
"maxcompute": ("kamidso.engine.ext.datasource.rdbms.conn_maxcompute", "MaxComputeParameters"),
|
|
"spark": ("kamidso.engine.ext.datasource.conn_spark", "SparkParameters"),
|
|
"neo4j": ("kamidso.engine.ext.datasource.conn_neo4j", "Neo4jParameters"),
|
|
"mongodb": ("kamidso.engine.ext.datasource.conn_mongodb", "MongoDBParameters"),
|
|
}
|
|
module_name, class_name = mapping[ds_type]
|
|
param_cls = getattr(importlib.import_module(module_name), class_name)
|
|
params = param_cls.from_dict(request_params, ignore_extra_fields=True)
|
|
params.create_connector()
|
|
|
|
def _invalidate(self, *docs: Dict[str, Any]) -> None:
|
|
try:
|
|
from kamidso.engine.core._private.config import Config
|
|
from kamidso.engine.serve.datasource.manages.connector_manager import ConnectorManager
|
|
except Exception:
|
|
return
|
|
system_app = getattr(Config(), "SYSTEM_APP", None)
|
|
if system_app is None:
|
|
return
|
|
try:
|
|
manager = ConnectorManager.get_instance(system_app)
|
|
except Exception:
|
|
return
|
|
names = set()
|
|
for doc in docs:
|
|
names.update(
|
|
{
|
|
str(doc.get("_id") or ""),
|
|
str(doc.get("name") or ""),
|
|
str(doc.get("db_name") or ""),
|
|
}
|
|
)
|
|
for name in names:
|
|
if name:
|
|
manager.invalidate_connector(name)
|