diff --git a/kamidso/native_connectors/runtime/__init__.py b/kamidso/native_connectors/runtime/__init__.py index e69de29..8b13789 100644 --- a/kamidso/native_connectors/runtime/__init__.py +++ b/kamidso/native_connectors/runtime/__init__.py @@ -0,0 +1 @@ + diff --git a/kamidso/resources/datasource_types.py b/kamidso/resources/datasource_types.py index 5c4ff90..434fdb6 100644 --- a/kamidso/resources/datasource_types.py +++ b/kamidso/resources/datasource_types.py @@ -5,16 +5,32 @@ from typing import Any, Dict, List, Optional from kamidso.contracts.errors import KamidsoError, KamidsoErrorCode from kamidso.resources.sqlite_path import resolve_sqlite_path +# Canonical ids match DB-GPT connector __type__ values, plus MariaDB (MySQL wire). SUPPORTED_TYPES = ( "mysql", + "mariadb", + "oceanbase", "postgresql", - "oracle", "mssql", + "oracle", "sqlite", + "duckdb", + "clickhouse", + "starrocks", + "doris", + "hive", + "gaussdb", + "openGauss", + "vertica", + "maxcompute", + "spark", "neo4j", "mongodb", ) +_FILE_TYPES = frozenset({"sqlite", "duckdb", "spark"}) +_PG_SCHEMA_TYPES = frozenset({"postgresql", "gaussdb", "openGauss"}) + _TYPE_META: Dict[str, Dict[str, Any]] = { "mysql": { "label": "MySQL", @@ -22,30 +38,102 @@ _TYPE_META: Dict[str, Dict[str, Any]] = { "default_port": 3306, "fields": ["host", "port", "user", "password", "database"], }, + "mariadb": { + "label": "MariaDB", + "kind": "sql", + "default_port": 3306, + "fields": ["host", "port", "user", "password", "database"], + }, + "oceanbase": { + "label": "OceanBase", + "kind": "sql", + "default_port": 2881, + "fields": ["host", "port", "user", "password", "database"], + }, "postgresql": { "label": "PostgreSQL", "kind": "sql", "default_port": 5432, "fields": ["host", "port", "user", "password", "database", "schema"], }, - "oracle": { - "label": "Oracle", - "kind": "sql", - "default_port": 1521, - "fields": ["host", "port", "user", "password", "service_name", "sid"], - }, "mssql": { "label": "Microsoft SQL Server", "kind": "sql", "default_port": 1433, "fields": ["host", "port", "user", "password", "database"], }, + "oracle": { + "label": "Oracle", + "kind": "sql", + "default_port": 1521, + "fields": ["host", "port", "user", "password", "service_name", "sid"], + }, "sqlite": { "label": "SQLite", "kind": "file", "default_port": None, "fields": ["path"], }, + "duckdb": { + "label": "DuckDB", + "kind": "file", + "default_port": None, + "fields": ["path"], + }, + "clickhouse": { + "label": "ClickHouse", + "kind": "sql", + "default_port": 8123, + "fields": ["host", "port", "user", "password", "database"], + }, + "starrocks": { + "label": "StarRocks", + "kind": "sql", + "default_port": 9030, + "fields": ["host", "port", "user", "password", "database"], + }, + "doris": { + "label": "Apache Doris", + "kind": "sql", + "default_port": 9030, + "fields": ["host", "port", "user", "password", "database"], + }, + "hive": { + "label": "Apache Hive", + "kind": "sql", + "default_port": 10000, + "fields": ["host", "port", "user", "password", "database"], + }, + "gaussdb": { + "label": "GaussDB", + "kind": "sql", + "default_port": 5432, + "fields": ["host", "port", "user", "password", "database", "schema"], + }, + "openGauss": { + "label": "openGauss", + "kind": "sql", + "default_port": 5432, + "fields": ["host", "port", "user", "password", "database", "schema"], + }, + "vertica": { + "label": "Vertica", + "kind": "sql", + "default_port": 5433, + "fields": ["host", "port", "user", "password", "database"], + }, + "maxcompute": { + "label": "MaxCompute", + "kind": "sql", + "default_port": None, + "fields": ["endpoint", "project", "access_id", "access_key"], + }, + "spark": { + "label": "Apache Spark", + "kind": "file", + "default_port": None, + "fields": ["path"], + }, "neo4j": { "label": "Neo4j", "kind": "graph", @@ -78,21 +166,36 @@ def catalog() -> List[Dict[str, Any]]: def normalize_type(ds_type: str) -> str: - key = (ds_type or "").strip().lower() + raw = (ds_type or "").strip() + key = raw.lower() aliases = { "postgres": "postgresql", "pgsql": "postgresql", "sqlserver": "mssql", "sql server": "mssql", "mongo": "mongodb", + "maria": "mariadb", + "opengauss": "openGauss", + "open_gauss": "openGauss", + "click house": "clickhouse", + "max compute": "maxcompute", + "odps": "maxcompute", } - key = aliases.get(key, key) - if key not in SUPPORTED_TYPES: + canonical = aliases.get(key, raw if raw in SUPPORTED_TYPES else key) + if canonical not in SUPPORTED_TYPES: raise KamidsoError( KamidsoErrorCode.VALIDATION_ERROR, details=f"Unsupported datasource type: {ds_type}", ) - return key + return canonical + + +def connector_engine_type(ds_type: str) -> str: + """Type string the DB-GPT ConnectorManager understands.""" + normalized = normalize_type(ds_type) + if normalized == "mariadb": + return "mysql" + return normalized def _text(config: Dict[str, Any], *keys: str, default: str = "") -> str: @@ -124,12 +227,41 @@ def normalize_params( ds_type = normalize_type(ds_type) config = dict(config or {}) meta = _TYPE_META[ds_type] - pwd = password if password is not None else _text(config, "password") - if ds_type == "sqlite": + pwd = password if password is not None else _text(config, "password", "access_key") + + if ds_type in _FILE_TYPES: path = _text(config, "path", "db_path", "file_path") if not path: - raise KamidsoError(KamidsoErrorCode.VALIDATION_ERROR, details="SQLite path is required") - return {"type": ds_type, "path": resolve_sqlite_path(path), "password": ""} + raise KamidsoError( + KamidsoErrorCode.VALIDATION_ERROR, + details="File path is required", + ) + if ds_type == "sqlite": + path = resolve_sqlite_path(path) + return {"type": ds_type, "path": path, "password": ""} + + if ds_type == "maxcompute": + endpoint = _text(config, "endpoint", "host") + project = _text(config, "project", "database") + if not endpoint: + raise KamidsoError( + KamidsoErrorCode.VALIDATION_ERROR, details="Endpoint is required" + ) + if not project: + raise KamidsoError( + KamidsoErrorCode.VALIDATION_ERROR, details="Project is required" + ) + return { + "type": ds_type, + "endpoint": endpoint, + "project": project, + "access_id": _text(config, "access_id", "user"), + "access_key": pwd, + "host": endpoint, + "database": project, + "user": _text(config, "access_id", "user"), + "password": pwd, + } host = _text(config, "host", "db_host") if not host: @@ -137,6 +269,8 @@ def normalize_params( database = _text(config, "database", "db_name") if ds_type == "neo4j" and not database: database = "neo4j" + if ds_type == "hive" and not database: + database = "default" if ds_type == "oracle": service_name = _text(config, "service_name") sid = _text(config, "sid") @@ -158,7 +292,7 @@ def normalize_params( "password": pwd, "database": database, } - if ds_type == "postgresql": + if ds_type in _PG_SCHEMA_TYPES: params["schema"] = _text(config, "schema") or "public" if ds_type == "oracle": params["service_name"] = _text(config, "service_name") @@ -169,15 +303,22 @@ def normalize_params( def public_config(params: Dict[str, Any]) -> Dict[str, Any]: - hidden = {"password", "type"} + hidden = {"password", "type", "access_key"} return {key: value for key, value in params.items() if key not in hidden and value not in (None, "")} def connector_params(params: Dict[str, Any]) -> Dict[str, Any]: """Shape expected by DB-GPT DatasourceCreateRequest.params.""" ds_type = params["type"] - if ds_type == "sqlite": + if ds_type in _FILE_TYPES: return {"path": params["path"]} + if ds_type == "maxcompute": + return { + "project": params["project"], + "access_id": params.get("access_id") or "", + "access_key": params.get("access_key") or params.get("password") or "", + "endpoint": params["endpoint"], + } payload: Dict[str, Any] = { "host": params["host"], "port": params["port"], @@ -185,11 +326,14 @@ def connector_params(params: Dict[str, Any]) -> Dict[str, Any]: "password": params.get("password") or "", "database": params["database"], } - if ds_type == "postgresql": + if ds_type in _PG_SCHEMA_TYPES: payload["schema"] = params.get("schema") or "public" if ds_type == "oracle": payload["service_name"] = params.get("service_name") or None payload["sid"] = params.get("sid") or None if ds_type == "mongodb": payload["auth_source"] = params.get("auth_source") or "admin" + if ds_type == "hive": + payload["username"] = params.get("user") or "" + payload.pop("user", None) return payload diff --git a/kamidso/resources/datasources.py b/kamidso/resources/datasources.py index 435e2c4..39d5d25 100644 --- a/kamidso/resources/datasources.py +++ b/kamidso/resources/datasources.py @@ -8,6 +8,7 @@ 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, @@ -150,26 +151,26 @@ class DatasourceService: ) -> Dict[str, Any]: ds_type = params["type"] ext: Dict[str, Any] = {} - for key in ("schema", "service_name", "sid", "auth_source"): + 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 "" - if ds_type == "sqlite": + 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"sqlite_{stem.split('.')[0] or 'db'}" + 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 "", + "db_host": params.get("host") or params.get("endpoint") or "", "db_port": params.get("port") or 0, - "db_user": params.get("user") or "", - "db_pwd": params.get("password") or "", + "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, @@ -183,7 +184,7 @@ class DatasourceService: } def _connect(self, params: Dict[str, Any]) -> None: - ds_type = params["type"] + ds_type = connector_engine_type(params["type"]) request_params = connector_params(params) try: from kamidso.engine.serve.datasource.api.schemas import DatasourceCreateRequest @@ -214,6 +215,17 @@ class DatasourceService: "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"), } diff --git a/tests/kamidso/test_datasources.py b/tests/kamidso/test_datasources.py index 9538894..6e6b9ac 100644 --- a/tests/kamidso/test_datasources.py +++ b/tests/kamidso/test_datasources.py @@ -5,7 +5,13 @@ import pytest 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, normalize_params, normalize_type +from kamidso.resources.datasource_types import ( + catalog, + connector_engine_type, + connector_params, + normalize_params, + normalize_type, +) from kamidso.resources.datasources import DatasourceService from kamidso.security.secrets import SecretStore @@ -27,7 +33,54 @@ def _service() -> DatasourceService: def test_catalog_includes_requested_engines(): ids = {item["id"] for item in catalog()} - assert ids == {"mysql", "postgresql", "oracle", "mssql", "sqlite", "neo4j", "mongodb"} + assert { + "mysql", + "mariadb", + "oceanbase", + "postgresql", + "mssql", + "oracle", + "sqlite", + "duckdb", + "clickhouse", + "starrocks", + "doris", + "hive", + "gaussdb", + "openGauss", + "vertica", + "maxcompute", + "spark", + "neo4j", + "mongodb", + }.issubset(ids) + + +def test_aliases_and_special_params(): + assert normalize_type("MariaDB") == "mariadb" + assert normalize_type("opengauss") == "openGauss" + duck = normalize_params("duckdb", {"path": "C:/data/sales.duckdb"}) + assert duck["type"] == "duckdb" + assert duck["path"].endswith("sales.duckdb") + maxc = normalize_params( + "maxcompute", + { + "endpoint": "https://service.cn-hangzhou.maxcompute.aliyun.com/api", + "project": "sales", + "access_id": "id", + }, + password="secret", + ) + assert maxc["project"] == "sales" + assert maxc["access_key"] == "secret" + assert connector_engine_type("mariadb") == "mysql" + hive = normalize_params("hive", {"host": "hiveserver", "user": "hive"}) + assert hive["database"] == "default" + hive_payload = connector_params(hive) + assert hive_payload["username"] == "hive" + assert "user" not in hive_payload + spark = connector_params(normalize_params("spark", {"path": "C:/data/sales.parquet"})) + assert spark["path"].endswith("sales.parquet") def test_normalize_mysql_params():