refactor(connections): remove ConnectionConfig, migrate mapper to SQL Lab
Backend: - Remove ConnectionConfig model, CRUD routes (connections.py), and tests - Remove ensure_connection_configs_table from database.py - Migrate MapperPlugin from psycopg2 direct PostgreSQL to SupersetSqlLabExecutor - Add DatasetMapper.get_sqllab_mappings() with default information_schema query - Update MapColumnsRequest: connection_id → database_id + sql_query - Fix pydantic_settings v2 deprecation (Field(env=...) → validation_alias) - Clean up app.py, routes/__init__.py, database.py imports Frontend: - Remove DatabaseConnectionsTab, ConnectionForm, ConnectionList components - Remove /settings/connections route page and Navbar link - Remove connectionService.js, connections i18n files - Update MapperTool.svelte: postgres → sqllab source - Update Map Columns modal: database selector + SQL query instead of connection_id - Clean up settings page tabs, nav links, test mocks - Clean up i18n keys (settings, nav, mapper) Tests: 25/25 pass (14 030-feature + 11 route tests) Build: succeeds
This commit is contained in:
@@ -12,7 +12,6 @@
|
||||
# @RELATION DEPENDS_ON -> [PluginsRouter]
|
||||
# @RELATION DEPENDS_ON -> [TasksRouter]
|
||||
# @RELATION DEPENDS_ON -> [SettingsRouter]
|
||||
# @RELATION DEPENDS_ON -> [ConnectionsRouter]
|
||||
# @RELATION DEPENDS_ON -> [ReportsRouter]
|
||||
# @RELATION DEPENDS_ON -> [LlmRoutes]
|
||||
# @SIDE_EFFECT: Registers route group imports via __getattr__
|
||||
@@ -22,7 +21,6 @@ __all__ = [
|
||||
"assistant",
|
||||
"clean_release",
|
||||
"clean_release_v2",
|
||||
"connections",
|
||||
"dashboards",
|
||||
"dataset_review",
|
||||
"datasets",
|
||||
|
||||
@@ -1,5 +1,11 @@
|
||||
# #region RoutesTestsConftest [TYPE Module] [C:1] [SEMANTICS test, fixture, mock, conftest]
|
||||
# @BRIEF Shared low-fidelity test doubles for API route test modules.
|
||||
|
||||
# Set required env vars before any app imports
|
||||
import os
|
||||
os.environ.setdefault("SECRET_KEY", "test-secret-key-for-testing")
|
||||
os.environ.setdefault("AUTH_DATABASE_URL", "sqlite:///:memory:test_auth")
|
||||
os.environ.setdefault("DEV_MODE", "true")
|
||||
class FakeQuery:
|
||||
"""Shared chainable query stub for route tests.
|
||||
WARNING: filter() is predicate-blind — all ownership and permission filters are
|
||||
|
||||
@@ -1,69 +0,0 @@
|
||||
# #region ConnectionsRoutesTests [TYPE Module] [C:3] [SEMANTICS test, connection, route, crud, bootstrap]
|
||||
# @BRIEF Verifies connection routes bootstrap their table before CRUD access.
|
||||
# @LAYER: API
|
||||
# @RELATION DEPENDS_ON -> ConnectionsRouter
|
||||
import asyncio
|
||||
import os
|
||||
from pathlib import Path
|
||||
import pytest
|
||||
import sys
|
||||
|
||||
from sqlalchemy import create_engine, inspect
|
||||
from sqlalchemy.orm import sessionmaker
|
||||
from sqlalchemy.pool import StaticPool
|
||||
|
||||
# Force SQLite in-memory for database module imports.
|
||||
# @SIDE_EFFECT_WARNING: os.environ mutation at module import time — no teardown. This bleeds into all subsequently collected tests. Migrate to pytest.fixture(autouse=True) with monkeypatch.setenv.
|
||||
os.environ["DATABASE_URL"] = "sqlite:///:memory:"
|
||||
# @SIDE_EFFECT_WARNING: os.environ mutation at module import time — no teardown. This bleeds into all subsequently collected tests. Migrate to pytest.fixture(autouse=True) with monkeypatch.setenv.
|
||||
os.environ["TASKS_DATABASE_URL"] = "sqlite:///:memory:"
|
||||
# @SIDE_EFFECT_WARNING: os.environ mutation at module import time — no teardown. This bleeds into all subsequently collected tests. Migrate to pytest.fixture(autouse=True) with monkeypatch.setenv.
|
||||
os.environ["AUTH_DATABASE_URL"] = "sqlite:///:memory:"
|
||||
# @SIDE_EFFECT_WARNING: os.environ mutation at module import time — no teardown. This bleeds into all subsequently collected tests. Migrate to pytest.fixture(autouse=True) with monkeypatch.setenv.
|
||||
os.environ["ENVIRONMENT"] = "testing"
|
||||
backend_dir = str(Path(__file__).parent.parent.parent.parent.resolve())
|
||||
if backend_dir not in sys.path:
|
||||
sys.path.insert(0, backend_dir)
|
||||
@pytest.fixture
|
||||
def db_session():
|
||||
engine = create_engine(
|
||||
"sqlite:///:memory:",
|
||||
connect_args={"check_same_thread": False},
|
||||
poolclass=StaticPool,
|
||||
)
|
||||
session = sessionmaker(bind=engine)()
|
||||
try:
|
||||
yield session
|
||||
finally:
|
||||
session.close()
|
||||
# #region test_list_connections_bootstraps_missing_table [TYPE Function]
|
||||
# @RELATION BINDS_TO -> ConnectionsRoutesTests
|
||||
# @BRIEF Ensure listing connections auto-creates missing table and returns empty payload.
|
||||
def test_list_connections_bootstraps_missing_table(db_session):
|
||||
from src.api.routes.connections import list_connections
|
||||
result = asyncio.run(list_connections(db=db_session))
|
||||
inspector = inspect(db_session.get_bind())
|
||||
assert result == []
|
||||
assert "connection_configs" in inspector.get_table_names()
|
||||
# #endregion test_list_connections_bootstraps_missing_table
|
||||
# #region test_create_connection_bootstraps_missing_table [TYPE Function]
|
||||
# @RELATION BINDS_TO -> ConnectionsRoutesTests
|
||||
# @BRIEF Ensure connection creation bootstraps table and persists returned connection fields.
|
||||
def test_create_connection_bootstraps_missing_table(db_session):
|
||||
from src.api.routes.connections import ConnectionCreate, create_connection
|
||||
payload = ConnectionCreate(
|
||||
name="Analytics Warehouse",
|
||||
type="postgres",
|
||||
host="warehouse.internal",
|
||||
port=5432,
|
||||
database="analytics",
|
||||
username="reporter",
|
||||
password="secret",
|
||||
)
|
||||
created = asyncio.run(create_connection(connection=payload, db=db_session))
|
||||
inspector = inspect(db_session.get_bind())
|
||||
assert created.name == "Analytics Warehouse"
|
||||
assert created.host == "warehouse.internal"
|
||||
assert "connection_configs" in inspector.get_table_names()
|
||||
# #endregion test_create_connection_bootstraps_missing_table
|
||||
# #endregion ConnectionsRoutesTests
|
||||
@@ -132,7 +132,7 @@ def test_get_datasets_invalid_pagination(mock_deps):
|
||||
# @RELATION BINDS_TO -> [DatasetsApiTests:Module]
|
||||
# @BRIEF Validate map-columns request creates an async mapping task and returns its identifier.
|
||||
# @TEST: POST /api/datasets/map-columns creates mapping task
|
||||
# @PRE: Valid env_id, dataset_ids, source_type
|
||||
# @PRE: Valid env_id, dataset_ids, source_type (sqllab)
|
||||
# @POST: Returns task_id
|
||||
def test_map_columns_success(mock_deps):
|
||||
# Mock environment
|
||||
@@ -145,7 +145,7 @@ def test_map_columns_success(mock_deps):
|
||||
mock_deps["task"].create_task = AsyncMock(return_value=mock_task)
|
||||
response = client.post(
|
||||
"/api/datasets/map-columns",
|
||||
json={"env_id": "prod", "dataset_ids": [1, 2, 3], "source_type": "postgresql"},
|
||||
json={"env_id": "prod", "dataset_ids": [1, 2, 3], "source_type": "sqllab", "database_id": 1},
|
||||
)
|
||||
assert response.status_code == 200
|
||||
data = response.json()
|
||||
@@ -157,7 +157,7 @@ def test_map_columns_success(mock_deps):
|
||||
# @RELATION BINDS_TO -> [DatasetsApiTests:Module]
|
||||
# @BRIEF Validate map-columns rejects unsupported source types with a 400 contract response.
|
||||
# @TEST: POST /api/datasets/map-columns returns 400 for invalid source_type
|
||||
# @PRE: source_type is not 'postgresql' or 'xlsx'
|
||||
# @PRE: source_type is not 'sqllab' or 'xlsx'
|
||||
# @POST: Returns 400 error
|
||||
def test_map_columns_invalid_source_type(mock_deps):
|
||||
response = client.post(
|
||||
@@ -165,7 +165,7 @@ def test_map_columns_invalid_source_type(mock_deps):
|
||||
json={"env_id": "prod", "dataset_ids": [1], "source_type": "invalid"},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
assert "Source type must be 'postgresql' or 'xlsx'" in response.json()["detail"]
|
||||
assert "Source type must be 'sqllab' or 'xlsx'" in response.json()["detail"]
|
||||
# #endregion test_map_columns_invalid_source_type
|
||||
# #region test_generate_docs_success [TYPE Function]
|
||||
# @RELATION BINDS_TO -> [DatasetsApiTests:Module]
|
||||
@@ -202,11 +202,24 @@ def test_map_columns_empty_ids(mock_deps):
|
||||
"""@PRE: dataset_ids must be non-empty."""
|
||||
response = client.post(
|
||||
"/api/datasets/map-columns",
|
||||
json={"env_id": "prod", "dataset_ids": [], "source_type": "postgresql"},
|
||||
json={"env_id": "prod", "dataset_ids": [], "source_type": "sqllab", "database_id": 1},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
assert "At least one dataset ID must be provided" in response.json()["detail"]
|
||||
# #endregion test_map_columns_empty_ids
|
||||
# #region test_map_columns_missing_database_id [TYPE Function]
|
||||
# @RELATION BINDS_TO -> [DatasetsApiTests:Module]
|
||||
# @BRIEF Validate map-columns rejects sqllab source without database_id.
|
||||
# @TEST: POST /api/datasets/map-columns returns 400 for sqllab without database_id
|
||||
# @POST: Returns 400 error
|
||||
def test_map_columns_missing_database_id(mock_deps):
|
||||
response = client.post(
|
||||
"/api/datasets/map-columns",
|
||||
json={"env_id": "prod", "dataset_ids": [1], "source_type": "sqllab"},
|
||||
)
|
||||
assert response.status_code == 400
|
||||
assert "database_id is required" in response.json()["detail"]
|
||||
# #endregion test_map_columns_missing_database_id
|
||||
# #region test_generate_docs_empty_ids [TYPE Function]
|
||||
# @RELATION BINDS_TO -> [DatasetsApiTests:Module]
|
||||
# @BRIEF Validate generate-docs rejects empty dataset identifier lists.
|
||||
|
||||
@@ -1,141 +0,0 @@
|
||||
# #region ConnectionsRouter [C:3] [TYPE Module] [SEMANTICS fastapi, connection, api, search]
|
||||
# @BRIEF Defines the FastAPI router for managing external database connections.
|
||||
# @LAYER: API
|
||||
# @RELATION DEPENDS_ON -> [get_db]
|
||||
# @RELATION DEPENDS_ON -> [ConnectionConfig]
|
||||
|
||||
from datetime import datetime
|
||||
|
||||
from fastapi import APIRouter, Depends, HTTPException, status
|
||||
from pydantic import BaseModel
|
||||
from sqlalchemy.orm import Session
|
||||
|
||||
from ...core.database import ensure_connection_configs_table, get_db
|
||||
from ...core.logger import belief_scope, logger
|
||||
from ...models.connection import ConnectionConfig
|
||||
|
||||
router = APIRouter()
|
||||
|
||||
|
||||
# #region _ensure_connections_schema [C:3] [TYPE Function]
|
||||
# @BRIEF Ensures the connection_configs table exists before CRUD access.
|
||||
# @PRE: db is an active SQLAlchemy session.
|
||||
# @POST: The current bind can safely query ConnectionConfig.
|
||||
# @RELATION CALLS -> ensure_connection_configs_table
|
||||
def _ensure_connections_schema(db: Session):
|
||||
with belief_scope("ConnectionsRouter.ensure_schema"):
|
||||
ensure_connection_configs_table(db.get_bind())
|
||||
|
||||
|
||||
# #endregion _ensure_connections_schema
|
||||
|
||||
|
||||
# #region ConnectionSchema [C:3] [TYPE Class]
|
||||
# @BRIEF Pydantic model for connection response.
|
||||
# @RELATION BINDS_TO -> ConnectionConfig
|
||||
class ConnectionSchema(BaseModel):
|
||||
id: str
|
||||
name: str
|
||||
type: str
|
||||
host: str | None = None
|
||||
port: int | None = None
|
||||
database: str | None = None
|
||||
username: str | None = None
|
||||
created_at: datetime
|
||||
|
||||
class Config:
|
||||
orm_mode = True
|
||||
|
||||
|
||||
# #endregion ConnectionSchema
|
||||
|
||||
|
||||
# #region ConnectionCreate [C:3] [TYPE Class]
|
||||
# @BRIEF Pydantic model for creating a connection.
|
||||
# @RELATION BINDS_TO -> ConnectionConfig
|
||||
class ConnectionCreate(BaseModel):
|
||||
name: str
|
||||
type: str
|
||||
host: str | None = None
|
||||
port: int | None = None
|
||||
database: str | None = None
|
||||
username: str | None = None
|
||||
password: str | None = None
|
||||
|
||||
|
||||
# #endregion ConnectionCreate
|
||||
|
||||
|
||||
# #region list_connections [C:3] [TYPE Function]
|
||||
# @BRIEF Lists all saved connections.
|
||||
# @PRE: Database session is active.
|
||||
# @POST: Returns list of connection configs.
|
||||
# @RELATION CALLS -> _ensure_connections_schema
|
||||
# @RELATION DEPENDS_ON -> ConnectionConfig
|
||||
@router.get("", response_model=list[ConnectionSchema])
|
||||
async def list_connections(db: Session = Depends(get_db)):
|
||||
with belief_scope("ConnectionsRouter.list_connections"):
|
||||
_ensure_connections_schema(db)
|
||||
connections = db.query(ConnectionConfig).all()
|
||||
return connections
|
||||
|
||||
|
||||
# #endregion list_connections
|
||||
|
||||
|
||||
# #region create_connection [C:3] [TYPE Function]
|
||||
# @BRIEF Creates a new connection configuration.
|
||||
# @PRE: Connection name is unique.
|
||||
# @POST: Connection is saved to DB.
|
||||
# @RELATION CALLS -> _ensure_connections_schema
|
||||
# @RELATION DEPENDS_ON -> ConnectionConfig
|
||||
@router.post("", response_model=ConnectionSchema, status_code=status.HTTP_201_CREATED)
|
||||
async def create_connection(
|
||||
connection: ConnectionCreate, db: Session = Depends(get_db)
|
||||
):
|
||||
with belief_scope("ConnectionsRouter.create_connection", f"name={connection.name}"):
|
||||
_ensure_connections_schema(db)
|
||||
db_connection = ConnectionConfig(**connection.dict())
|
||||
db.add(db_connection)
|
||||
db.commit()
|
||||
db.refresh(db_connection)
|
||||
logger.info(
|
||||
f"[ConnectionsRouter.create_connection][Success] Created connection {db_connection.id}"
|
||||
)
|
||||
return db_connection
|
||||
|
||||
|
||||
# #endregion create_connection
|
||||
|
||||
|
||||
# #region delete_connection [C:3] [TYPE Function]
|
||||
# @BRIEF Deletes a connection configuration.
|
||||
# @PRE: Connection ID exists.
|
||||
# @POST: Connection is removed from DB.
|
||||
# @RELATION CALLS -> _ensure_connections_schema
|
||||
# @RELATION DEPENDS_ON -> ConnectionConfig
|
||||
@router.delete("/{connection_id}", status_code=status.HTTP_204_NO_CONTENT)
|
||||
async def delete_connection(connection_id: str, db: Session = Depends(get_db)):
|
||||
with belief_scope("ConnectionsRouter.delete_connection", f"id={connection_id}"):
|
||||
_ensure_connections_schema(db)
|
||||
db_connection = (
|
||||
db.query(ConnectionConfig)
|
||||
.filter(ConnectionConfig.id == connection_id)
|
||||
.first()
|
||||
)
|
||||
if not db_connection:
|
||||
logger.error(
|
||||
f"[ConnectionsRouter.delete_connection][State] Connection {connection_id} not found"
|
||||
)
|
||||
raise HTTPException(status_code=404, detail="Connection not found")
|
||||
db.delete(db_connection)
|
||||
db.commit()
|
||||
logger.info(
|
||||
f"[ConnectionsRouter.delete_connection][Success] Deleted connection {connection_id}"
|
||||
)
|
||||
return
|
||||
|
||||
|
||||
# #endregion delete_connection
|
||||
|
||||
# #endregion ConnectionsRouter
|
||||
@@ -316,8 +316,9 @@ async def get_datasets(
|
||||
class MapColumnsRequest(BaseModel):
|
||||
env_id: str = Field(..., description="Environment ID")
|
||||
dataset_ids: list[int] = Field(..., description="List of dataset IDs to map")
|
||||
source_type: str = Field(..., description="Source type: 'postgresql' or 'xlsx'")
|
||||
connection_id: str | None = Field(None, description="Connection ID for PostgreSQL source")
|
||||
source_type: str = Field(..., description="Source type: 'sqllab' or 'xlsx'")
|
||||
database_id: int | None = Field(None, description="Superset database ID for SQL Lab source")
|
||||
sql_query: str | None = Field(None, description="Custom SQL query for SQL Lab (optional, defaults to information_schema.columns query)")
|
||||
file_data: str | None = Field(None, description="File path or data for XLSX source")
|
||||
# #endregion MapColumnsRequest
|
||||
|
||||
@@ -344,9 +345,13 @@ async def map_columns(
|
||||
raise HTTPException(status_code=400, detail="At least one dataset ID must be provided")
|
||||
|
||||
# Validate source type
|
||||
if request.source_type not in ['postgresql', 'xlsx']:
|
||||
if request.source_type not in ['sqllab', 'xlsx']:
|
||||
logger.error(f"[map_columns][Coherence:Failed] Invalid source type: {request.source_type}")
|
||||
raise HTTPException(status_code=400, detail="Source type must be 'postgresql' or 'xlsx'")
|
||||
raise HTTPException(status_code=400, detail="Source type must be 'sqllab' or 'xlsx'")
|
||||
|
||||
# Validate database_id for sqllab source
|
||||
if request.source_type == 'sqllab' and not request.database_id:
|
||||
raise HTTPException(status_code=400, detail="database_id is required for 'sqllab' source type")
|
||||
|
||||
# Validate environment exists
|
||||
environments = config_manager.get_environments()
|
||||
@@ -361,8 +366,9 @@ async def map_columns(
|
||||
task_params = {
|
||||
'env': request.env_id,
|
||||
'dataset_id': request.dataset_ids[0] if request.dataset_ids else None,
|
||||
'source': request.source_type,
|
||||
'connection_id': request.connection_id,
|
||||
'source': 'sqllab' if request.source_type == 'sqllab' else 'excel',
|
||||
'database_id': request.database_id,
|
||||
'sql_query': request.sql_query,
|
||||
'file_data': request.file_data
|
||||
}
|
||||
|
||||
|
||||
@@ -28,7 +28,6 @@ from .api.routes import (
|
||||
assistant,
|
||||
clean_release,
|
||||
clean_release_v2,
|
||||
connections,
|
||||
dashboards,
|
||||
dataset_review,
|
||||
datasets,
|
||||
@@ -215,7 +214,6 @@ app.add_middleware(TraceContextMiddleware)
|
||||
# @RELATION DEPENDS_ON -> [PluginsRouter]
|
||||
# @RELATION DEPENDS_ON -> [TasksRouter]
|
||||
# @RELATION DEPENDS_ON -> [SettingsRouter]
|
||||
# @RELATION DEPENDS_ON -> [ConnectionsRouter]
|
||||
# @RELATION DEPENDS_ON -> [ReportsRouter]
|
||||
# @RELATION DEPENDS_ON -> [LlmRoutes]
|
||||
# @RELATION DEPENDS_ON -> [CleanReleaseV2Api]
|
||||
@@ -225,9 +223,6 @@ app.include_router(admin.router)
|
||||
app.include_router(plugins.router, prefix="/api/plugins", tags=["Plugins"])
|
||||
app.include_router(tasks.router, prefix="/api/tasks", tags=["Tasks"])
|
||||
app.include_router(settings.router, prefix="/api/settings", tags=["Settings"])
|
||||
app.include_router(
|
||||
connections.router, prefix="/api/settings/connections", tags=["Connections"]
|
||||
)
|
||||
app.include_router(environments.router, tags=["Environments"])
|
||||
app.include_router(mappings.router, prefix="/api/mappings", tags=["Mappings"])
|
||||
app.include_router(migration.router)
|
||||
|
||||
@@ -14,7 +14,7 @@
|
||||
import os
|
||||
|
||||
from pydantic import Field, field_validator
|
||||
from pydantic_settings import BaseSettings
|
||||
from pydantic_settings import BaseSettings, SettingsConfigDict
|
||||
|
||||
|
||||
# #region AuthConfig [TYPE Class]
|
||||
@@ -23,19 +23,21 @@ from pydantic_settings import BaseSettings
|
||||
# @POST: Returns a configuration object with validated settings.
|
||||
# @RELATION INHERITS -> pydantic_settings.BaseSettings
|
||||
class AuthConfig(BaseSettings):
|
||||
model_config = SettingsConfigDict(env_file=".env", extra="ignore")
|
||||
|
||||
# JWT Settings
|
||||
SECRET_KEY: str = Field(default="", env="AUTH_SECRET_KEY")
|
||||
SECRET_KEY: str = Field(default="", validation_alias="AUTH_SECRET_KEY")
|
||||
ALGORITHM: str = "HS256"
|
||||
ACCESS_TOKEN_EXPIRE_MINUTES: int = 480
|
||||
REFRESH_TOKEN_EXPIRE_DAYS: int = 7
|
||||
|
||||
# Database Settings
|
||||
AUTH_DATABASE_URL: str = Field(default="", env="AUTH_DATABASE_URL")
|
||||
AUTH_DATABASE_URL: str = Field(default="", validation_alias="AUTH_DATABASE_URL")
|
||||
|
||||
# ADFS Settings
|
||||
ADFS_CLIENT_ID: str = Field(default="", env="ADFS_CLIENT_ID")
|
||||
ADFS_CLIENT_SECRET: str = Field(default="", env="ADFS_CLIENT_SECRET")
|
||||
ADFS_METADATA_URL: str = Field(default="", env="ADFS_METADATA_URL")
|
||||
ADFS_CLIENT_ID: str = Field(default="", validation_alias="ADFS_CLIENT_ID")
|
||||
ADFS_CLIENT_SECRET: str = Field(default="", validation_alias="ADFS_CLIENT_SECRET")
|
||||
ADFS_METADATA_URL: str = Field(default="", validation_alias="ADFS_METADATA_URL")
|
||||
|
||||
@field_validator("SECRET_KEY", mode="after")
|
||||
@classmethod
|
||||
@@ -60,9 +62,6 @@ class AuthConfig(BaseSettings):
|
||||
"Set it in .env or export it before starting the server."
|
||||
)
|
||||
|
||||
class Config:
|
||||
env_file = ".env"
|
||||
extra = "ignore"
|
||||
# #endregion AuthConfig
|
||||
|
||||
# #region auth_config [TYPE Variable]
|
||||
|
||||
@@ -4,7 +4,6 @@
|
||||
# @LAYER: Infrastructure
|
||||
# @RELATION DEPENDS_ON -> [MappingModels]
|
||||
# @RELATION DEPENDS_ON -> [auth_config]
|
||||
# @RELATION DEPENDS_ON -> [ConnectionConfig]
|
||||
#
|
||||
# @INVARIANT: A single engine instance is used for the entire application.
|
||||
|
||||
@@ -20,13 +19,11 @@ from ..models import (
|
||||
auth as _auth_models, # noqa: F401
|
||||
clean_release as _clean_release_models, # noqa: F401
|
||||
config as _config_models, # noqa: F401
|
||||
connection as _connection_models, # noqa: F401
|
||||
dataset_review as _dataset_review_models, # noqa: F401
|
||||
llm as _llm_models, # noqa: F401
|
||||
profile as _profile_models, # noqa: F401
|
||||
task as _task_models, # noqa: F401
|
||||
)
|
||||
from ..models.connection import ConnectionConfig
|
||||
from ..models.mapping import Base
|
||||
from .auth.config import auth_config
|
||||
from .logger import belief_scope, logger
|
||||
@@ -367,26 +364,6 @@ def _ensure_auth_users_columns(bind_engine):
|
||||
# #endregion _ensure_auth_users_columns
|
||||
|
||||
|
||||
# #region ensure_connection_configs_table [C:3] [TYPE Function]
|
||||
# @BRIEF Ensures the external connection registry table exists in the main database.
|
||||
# @PRE: bind_engine points to the application database.
|
||||
# @POST: connection_configs table exists without dropping existing data.
|
||||
# @RELATION DEPENDS_ON -> [ConnectionConfig]
|
||||
def ensure_connection_configs_table(bind_engine):
|
||||
with belief_scope("ensure_connection_configs_table"):
|
||||
try:
|
||||
ConnectionConfig.__table__.create(bind=bind_engine, checkfirst=True)
|
||||
except Exception as migration_error:
|
||||
logger.explore(
|
||||
"ConnectionConfig table ensure failed",
|
||||
extra={"src": "database", "error": str(migration_error)},
|
||||
)
|
||||
raise
|
||||
|
||||
|
||||
# #endregion ensure_connection_configs_table
|
||||
|
||||
|
||||
# #region _ensure_filter_source_enum_values [C:3] [TYPE Function]
|
||||
# @BRIEF Adds missing FilterSource enum values to the PostgreSQL native filtersource type.
|
||||
# @PRE: bind_engine points to application database with imported_filters table.
|
||||
@@ -686,7 +663,6 @@ def _ensure_dictionary_entries_columns(bind_engine):
|
||||
# @PRE: engine, tasks_engine and auth_engine are initialized.
|
||||
# @POST: Database tables created in all databases.
|
||||
# @SIDE_EFFECT: Creates physical database files if they don't exist.
|
||||
# @RELATION CALLS -> [ensure_connection_configs_table]
|
||||
# @RELATION CALLS -> [_ensure_filter_source_enum_values]
|
||||
# @RELATION CALLS -> [_ensure_dataset_review_session_columns]
|
||||
# @RELATION CALLS -> [_ensure_dictionary_entries_columns]
|
||||
@@ -700,7 +676,6 @@ def init_db():
|
||||
_ensure_user_dashboard_preferences_health_columns(engine)
|
||||
_ensure_git_server_configs_columns(engine)
|
||||
_ensure_auth_users_columns(auth_engine)
|
||||
ensure_connection_configs_table(engine)
|
||||
_ensure_filter_source_enum_values(engine)
|
||||
_ensure_dataset_review_session_columns(engine)
|
||||
_ensure_translation_jobs_columns(engine)
|
||||
|
||||
@@ -1,21 +1,20 @@
|
||||
# #region DatasetMapperModule [TYPE Module] [SEMANTICS dataset, superset, dataset-mapper]
|
||||
# #region DatasetMapperModule [TYPE Module] [SEMANTICS dataset, superset, dataset-mapper, sqllab]
|
||||
#
|
||||
# @BRIEF Этот модуль отвечает за обновление метаданных (verbose_map) в датасетах Superset, извлекая их из PostgreSQL или XLSX-файлов.
|
||||
# @BRIEF Этот модуль отвечает за обновление метаданных (verbose_name) в датасетах Superset,
|
||||
# извлекая их из Superset SQL Lab (любая БД, подключённая к Superset) или XLSX-файлов.
|
||||
# @LAYER: Domain
|
||||
# @RELATION DEPENDS_ON -> backend.core.superset_client
|
||||
# @RELATION DEPENDS_ON -> pandas
|
||||
# @RELATION DEPENDS_ON -> psycopg2
|
||||
# @PUBLIC_API: DatasetMapper
|
||||
from typing import Any
|
||||
|
||||
import pandas as pd # type: ignore
|
||||
import psycopg2 # type: ignore
|
||||
|
||||
from ..logger import belief_scope, logger as app_logger
|
||||
|
||||
|
||||
# #region DatasetMapper [TYPE Class]
|
||||
# @BRIEF Класс для меппинга и обновления verbose_map в датасетах Superset.
|
||||
# @BRIEF Класс для маппинга и обновления verbose_name в датасетах Superset.
|
||||
class DatasetMapper:
|
||||
# #region __init__ [TYPE Function]
|
||||
# @PURPOSE: Initializes the mapper.
|
||||
@@ -23,77 +22,78 @@ class DatasetMapper:
|
||||
def __init__(self):
|
||||
pass
|
||||
# #endregion __init__
|
||||
# #region get_postgres_comments [TYPE Function]
|
||||
# @PURPOSE: Извлекает комментарии к колонкам из системного каталога PostgreSQL.
|
||||
# @PRE: db_config должен содержать валидные параметры подключения (host, port, user, password, dbname).
|
||||
# @PRE: table_name и table_schema должны быть строками.
|
||||
# @POST: Возвращается словарь, где ключи - имена колонок, значения - комментарии из БД.
|
||||
# @THROW: Exception - При ошибках подключения или выполнения запроса к БД.
|
||||
# @PARAM: db_config (Dict) - Конфигурация для подключения к БД.
|
||||
# @PARAM: table_name (str) - Имя таблицы.
|
||||
# @PARAM: table_schema (str) - Схема таблицы.
|
||||
# @RETURN: Dict[str, str] - Словарь с комментариями к колонкам.
|
||||
def get_postgres_comments(self, db_config: dict, table_name: str, table_schema: str) -> dict[str, str]:
|
||||
with belief_scope("Fetch comments from PostgreSQL"):
|
||||
app_logger.info("[get_postgres_comments][Enter] Fetching comments from PostgreSQL for %s.%s.", table_schema, table_name)
|
||||
query = f"""
|
||||
SELECT
|
||||
cols.column_name,
|
||||
CASE
|
||||
WHEN pg_catalog.col_description(
|
||||
(SELECT c.oid
|
||||
FROM pg_catalog.pg_class c
|
||||
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
|
||||
WHERE c.relname = cols.table_name
|
||||
AND n.nspname = cols.table_schema),
|
||||
cols.ordinal_position::int
|
||||
) LIKE '%|%' THEN
|
||||
split_part(
|
||||
pg_catalog.col_description(
|
||||
(SELECT c.oid
|
||||
FROM pg_catalog.pg_class c
|
||||
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
|
||||
WHERE c.relname = cols.table_name
|
||||
AND n.nspname = cols.table_schema),
|
||||
cols.ordinal_position::int
|
||||
),
|
||||
'|',
|
||||
1
|
||||
)
|
||||
ELSE
|
||||
pg_catalog.col_description(
|
||||
(SELECT c.oid
|
||||
FROM pg_catalog.pg_class c
|
||||
JOIN pg_catalog.pg_namespace n ON n.oid = c.relnamespace
|
||||
WHERE c.relname = cols.table_name
|
||||
AND n.nspname = cols.table_schema),
|
||||
cols.ordinal_position::int
|
||||
)
|
||||
END AS column_comment
|
||||
FROM
|
||||
information_schema.columns cols
|
||||
WHERE cols.table_catalog = '{db_config.get('dbname')}' AND cols.table_name = '{table_name}' AND cols.table_schema = '{table_schema}';
|
||||
"""
|
||||
comments = {}
|
||||
try:
|
||||
with psycopg2.connect(**db_config) as conn, conn.cursor() as cursor:
|
||||
cursor.execute(query)
|
||||
for row in cursor.fetchall():
|
||||
if row[1]:
|
||||
comments[row[0]] = row[1]
|
||||
app_logger.info("[get_postgres_comments][Success] Fetched %d comments.", len(comments))
|
||||
except Exception as e:
|
||||
app_logger.error("[get_postgres_comments][Failure] %s", e, exc_info=True)
|
||||
raise
|
||||
return comments
|
||||
# #endregion get_postgres_comments
|
||||
|
||||
# #region get_sqllab_mappings [TYPE Function]
|
||||
# @PURPOSE: Извлекает маппинги column_name -> verbose_name через SQL Lab Superset.
|
||||
# @PRE: sqllab_executor должен быть инициализирован с database_id.
|
||||
# @PRE: dataset_id должен существовать в Superset.
|
||||
# @POST: Возвращается словарь column_name -> verbose_name из результатов SQL-запроса.
|
||||
# Если sql_query не указан, строится запрос к information_schema.columns.
|
||||
# @PARAM: client (SupersetClient) - Авторизованный клиент Superset.
|
||||
# @PARAM: dataset_id (int) - ID датасета (для получения table_name/schema).
|
||||
# @PARAM: sqllab_executor (SupersetSqlLabExecutor) - Исполнитель SQL Lab.
|
||||
# @PARAM: database_id (int) - ID базы данных в Superset.
|
||||
# @PARAM: sql_query (Optional[str]) - Произвольный SQL-запрос (должен вернуть column_name + verbose_name).
|
||||
# @RETURN: Dict[str, str] - Словарь column_name -> verbose_name.
|
||||
def get_sqllab_mappings(
|
||||
self,
|
||||
client: Any,
|
||||
dataset_id: int,
|
||||
sqllab_executor: Any,
|
||||
database_id: int,
|
||||
sql_query: str | None = None,
|
||||
) -> dict[str, str]:
|
||||
with belief_scope("Fetch mappings via SQL Lab"):
|
||||
# Получаем датасет из Superset, чтобы узнать table_name и schema
|
||||
dataset_response = client.get_dataset(dataset_id)
|
||||
dataset_data = dataset_response.get("result", dataset_response)
|
||||
table_name = dataset_data.get("table_name")
|
||||
table_schema = dataset_data.get("schema") or "public"
|
||||
database_name = dataset_data.get("database", {}).get("database_name", "")
|
||||
|
||||
app_logger.info(
|
||||
"[get_sqllab_mappings] Dataset %d: %s.%s (db: %s)",
|
||||
dataset_id, table_schema, table_name, database_name
|
||||
)
|
||||
|
||||
# Если SQL-запрос не указан, строим дефолтный для information_schema
|
||||
if not sql_query:
|
||||
sql_query = (
|
||||
"SELECT column_name, pg_catalog.col_description(\n"
|
||||
" (quote_ident(table_schema) || '.' || quote_ident(table_name))::regclass::oid,\n"
|
||||
" ordinal_position\n"
|
||||
") AS verbose_name\n"
|
||||
f"FROM information_schema.columns\n"
|
||||
f"WHERE table_schema = '{table_schema}' AND table_name = '{table_name}';\n"
|
||||
)
|
||||
app_logger.info("[get_sqllab_mappings] Using default information_schema query for %s.%s", table_schema, table_name)
|
||||
else:
|
||||
app_logger.info("[get_sqllab_mappings] Using custom SQL query")
|
||||
|
||||
# Выполняем запрос через SQL Lab
|
||||
app_logger.info("[get_sqllab_mappings] Executing SQL Lab query on database %d...", database_id)
|
||||
result = sqllab_executor.execute_and_poll(sql_query, database_id)
|
||||
rows = result.get("results") or result.get("data") or result.get("result", [])
|
||||
app_logger.info("[get_sqllab_mappings] SQL Lab returned %d rows", len(rows))
|
||||
|
||||
mappings: dict[str, str] = {}
|
||||
for row in rows:
|
||||
col_name = row.get("column_name")
|
||||
verbose_name = row.get("verbose_name")
|
||||
if col_name and verbose_name:
|
||||
mappings[col_name] = str(verbose_name)
|
||||
|
||||
app_logger.info("[get_sqllab_mappings] Extracted %d mappings", len(mappings))
|
||||
return mappings
|
||||
# #endregion get_sqllab_mappings
|
||||
|
||||
# #region load_excel_mappings [TYPE Function]
|
||||
# @PURPOSE: Загружает меппинги 'column_name' -> 'column_comment' из XLSX файла.
|
||||
# @PURPOSE: Загружает маппинги column_name -> verbose_name из XLSX файла.
|
||||
# @PRE: file_path должен указывать на существующий XLSX файл.
|
||||
# @POST: Возвращается словарь с меппингами из файла.
|
||||
# @POST: Возвращается словарь с маппингами из файла.
|
||||
# @THROW: Exception - При ошибках чтения файла или парсинга.
|
||||
# @PARAM: file_path (str) - Путь к XLSX файлу.
|
||||
# @RETURN: Dict[str, str] - Словарь с меппингами.
|
||||
# @RETURN: Dict[str, str] - Словарь с маппингами.
|
||||
def load_excel_mappings(self, file_path: str) -> dict[str, str]:
|
||||
with belief_scope("Load mappings from Excel"):
|
||||
app_logger.info("[load_excel_mappings][Enter] Loading mappings from %s.", file_path)
|
||||
@@ -106,37 +106,48 @@ class DatasetMapper:
|
||||
app_logger.error("[load_excel_mappings][Failure] %s", e, exc_info=True)
|
||||
raise
|
||||
# #endregion load_excel_mappings
|
||||
|
||||
# #region run_mapping [TYPE Function]
|
||||
# @PURPOSE: Основная функция для выполнения меппинга и обновления verbose_map датасета в Superset.
|
||||
# @PURPOSE: Основная функция для выполнения маппинга и обновления verbose_name датасета в Superset.
|
||||
# @PRE: superset_client должен быть авторизован.
|
||||
# @PRE: dataset_id должен быть существующим ID в Superset.
|
||||
# @POST: Если найдены изменения, датасет в Superset обновлен через API.
|
||||
# @RELATION: CALLS -> self.get_postgres_comments
|
||||
# @RELATION: CALLS -> self.get_sqllab_mappings
|
||||
# @RELATION: CALLS -> self.load_excel_mappings
|
||||
# @RELATION: CALLS -> superset_client.get_dataset
|
||||
# @RELATION: CALLS -> superset_client.update_dataset
|
||||
# @PARAM: superset_client (Any) - Клиент Superset.
|
||||
# @PARAM: dataset_id (int) - ID датасета для обновления.
|
||||
# @PARAM: source (str) - Источник данных ('postgres', 'excel', 'both').
|
||||
# @PARAM: postgres_config (Optional[Dict]) - Конфигурация для подключения к PostgreSQL.
|
||||
# @PARAM: source (str) - Источник данных ('sqllab' или 'excel').
|
||||
# @PARAM: sqllab_executor (Optional[Any]) - Исполнитель SQL Lab (для sqllab source).
|
||||
# @PARAM: database_id (Optional[int]) - ID базы данных Superset (для sqllab source).
|
||||
# @PARAM: sql_query (Optional[str]) - Произвольный SQL-запрос (для sqllab source).
|
||||
# @PARAM: excel_path (Optional[str]) - Путь к XLSX файлу.
|
||||
# @PARAM: table_name (Optional[str]) - Имя таблицы в PostgreSQL.
|
||||
# @PARAM: table_schema (Optional[str]) - Схема таблицы в PostgreSQL.
|
||||
def run_mapping(self, superset_client: Any, dataset_id: int, source: str, postgres_config: dict | None = None, excel_path: str | None = None, table_name: str | None = None, table_schema: str | None = None):
|
||||
def run_mapping(
|
||||
self,
|
||||
superset_client: Any,
|
||||
dataset_id: int,
|
||||
source: str,
|
||||
sqllab_executor: Any = None,
|
||||
database_id: int | None = None,
|
||||
sql_query: str | None = None,
|
||||
excel_path: str | None = None,
|
||||
):
|
||||
with belief_scope(f"Run dataset mapping for ID {dataset_id}"):
|
||||
app_logger.info("[run_mapping][Enter] Starting dataset mapping for ID %d from source '%s'.", dataset_id, source)
|
||||
mappings: dict[str, str] = {}
|
||||
|
||||
try:
|
||||
if source in ['postgres', 'both']:
|
||||
assert postgres_config and table_name and table_schema, "Postgres config is required."
|
||||
mappings.update(self.get_postgres_comments(postgres_config, table_name, table_schema))
|
||||
if source in ['excel', 'both']:
|
||||
assert excel_path, "Excel path is required."
|
||||
if source == "sqllab":
|
||||
assert sqllab_executor and database_id, "sqllab_executor and database_id are required for sqllab source."
|
||||
mappings.update(self.get_sqllab_mappings(superset_client, dataset_id, sqllab_executor, database_id, sql_query))
|
||||
elif source == "excel":
|
||||
assert excel_path, "excel_path is required."
|
||||
mappings.update(self.load_excel_mappings(excel_path))
|
||||
if source not in ['postgres', 'excel', 'both']:
|
||||
else:
|
||||
app_logger.error("[run_mapping][Failure] Invalid source: %s.", source)
|
||||
return
|
||||
|
||||
dataset_response = superset_client.get_dataset(dataset_id)
|
||||
dataset_data = dataset_response['result']
|
||||
|
||||
@@ -171,6 +182,7 @@ class DatasetMapper:
|
||||
changes_made = True
|
||||
|
||||
updated_columns.append(new_column)
|
||||
|
||||
updated_metrics = []
|
||||
for metric in dataset_data.get("metrics", []):
|
||||
new_metric = {
|
||||
@@ -187,6 +199,7 @@ class DatasetMapper:
|
||||
"uuid": metric.get("uuid"),
|
||||
}
|
||||
updated_metrics.append({k: v for k, v in new_metric.items() if v is not None})
|
||||
|
||||
if changes_made:
|
||||
payload_for_update = {
|
||||
"database_id": dataset_data.get("database", {}).get("id"),
|
||||
@@ -222,4 +235,4 @@ class DatasetMapper:
|
||||
return
|
||||
# #endregion run_mapping
|
||||
# #endregion DatasetMapper
|
||||
# #endregion DatasetMapperModule
|
||||
# #endregion DatasetMapperModule
|
||||
@@ -1,34 +0,0 @@
|
||||
# #region ConnectionModels [C:1] [TYPE Module] [SEMANTICS sqlalchemy, connection, model, schema, connection-config]
|
||||
#
|
||||
# @BRIEF Defines the database schema for external database connection configurations.
|
||||
# @LAYER: Domain
|
||||
# @RELATION DEPENDS_ON -> sqlalchemy
|
||||
#
|
||||
# @INVARIANT: All primary keys are UUID strings.
|
||||
|
||||
import uuid
|
||||
|
||||
from sqlalchemy import Column, DateTime, Integer, String
|
||||
from sqlalchemy.sql import func
|
||||
|
||||
from .mapping import Base
|
||||
|
||||
|
||||
# #region ConnectionConfig [C:1] [TYPE Class]
|
||||
# @BRIEF Stores credentials for external databases used for column mapping.
|
||||
class ConnectionConfig(Base):
|
||||
__tablename__ = "connection_configs"
|
||||
|
||||
id = Column(String, primary_key=True, default=lambda: str(uuid.uuid4()))
|
||||
name = Column(String, nullable=False)
|
||||
type = Column(String, nullable=False) # e.g., "postgres"
|
||||
host = Column(String, nullable=True)
|
||||
port = Column(Integer, nullable=True)
|
||||
database = Column(String, nullable=True)
|
||||
username = Column(String, nullable=True)
|
||||
password = Column(String, nullable=True) # Encrypted/Obfuscated password
|
||||
created_at = Column(DateTime(timezone=True), server_default=func.now())
|
||||
updated_at = Column(DateTime(timezone=True), server_default=func.now(), onupdate=func.now())
|
||||
# #endregion ConnectionConfig
|
||||
|
||||
# #endregion ConnectionModels
|
||||
@@ -1,18 +1,16 @@
|
||||
# #region MapperPluginModule [TYPE Module] [SEMANTICS mapper, dataset, column, excel, mapping]
|
||||
# @BRIEF Implements a plugin for mapping dataset columns using external database connections or Excel files.
|
||||
# #region MapperPluginModule [TYPE Module] [SEMANTICS mapper, dataset, column, sqllab, excel, mapping]
|
||||
# @BRIEF Implements a plugin for mapping dataset columns using Superset SQL Lab or Excel files.
|
||||
# @LAYER: Plugins
|
||||
# @RELATION Inherits from PluginBase. Uses DatasetMapper from superset_tool.
|
||||
# @RELATION Inherits from PluginBase. Uses DatasetMapper and SupersetSqlLabExecutor.
|
||||
# @RELATION USES -> TaskContext
|
||||
|
||||
from typing import Any
|
||||
|
||||
from ..core.database import SessionLocal
|
||||
from ..core.logger import belief_scope, logger
|
||||
from ..core.plugin_base import PluginBase
|
||||
from ..core.superset_client import SupersetClient
|
||||
from ..core.task_manager.context import TaskContext
|
||||
from ..core.utils.dataset_mapper import DatasetMapper
|
||||
from ..models.connection import ConnectionConfig
|
||||
|
||||
|
||||
# #region MapperPlugin [TYPE Class]
|
||||
@@ -52,7 +50,7 @@ class MapperPlugin(PluginBase):
|
||||
# @RETURN: str - Plugin description.
|
||||
def description(self) -> str:
|
||||
with belief_scope("description"):
|
||||
return "Map dataset column verbose names using PostgreSQL comments or Excel files."
|
||||
return "Map dataset column verbose names using Superset SQL Lab or Excel files."
|
||||
# endregion description
|
||||
|
||||
@property
|
||||
@@ -98,24 +96,19 @@ class MapperPlugin(PluginBase):
|
||||
"source": {
|
||||
"type": "string",
|
||||
"title": "Mapping Source",
|
||||
"enum": ["postgres", "excel"],
|
||||
"default": "postgres"
|
||||
"enum": ["sqllab", "excel"],
|
||||
"default": "sqllab"
|
||||
},
|
||||
"connection_id": {
|
||||
"type": "string",
|
||||
"title": "Saved Connection",
|
||||
"description": "The ID of a saved database connection (for postgres source)."
|
||||
"database_id": {
|
||||
"type": "integer",
|
||||
"title": "Superset Database ID",
|
||||
"description": "Target database ID in Superset (for sqllab source)."
|
||||
},
|
||||
"table_name": {
|
||||
"sql_query": {
|
||||
"type": "string",
|
||||
"title": "Table Name",
|
||||
"description": "Target table name in PostgreSQL."
|
||||
},
|
||||
"table_schema": {
|
||||
"type": "string",
|
||||
"title": "Table Schema",
|
||||
"description": "Target table schema in PostgreSQL.",
|
||||
"default": "public"
|
||||
"title": "SQL Query",
|
||||
"description": "Custom SQL query returning column_name and verbose_name columns. Default: information_schema.columns.",
|
||||
"default": ""
|
||||
},
|
||||
"excel_path": {
|
||||
"type": "string",
|
||||
@@ -145,7 +138,6 @@ class MapperPlugin(PluginBase):
|
||||
|
||||
# Create sub-loggers for different components
|
||||
superset_log = log.with_source("superset_api") if context else log
|
||||
db_log = log.with_source("postgres") if context else log
|
||||
|
||||
if not env_name or dataset_id is None or not source:
|
||||
log.error("Missing required parameters: env, dataset_id, source")
|
||||
@@ -162,46 +154,37 @@ class MapperPlugin(PluginBase):
|
||||
client = SupersetClient(env_config)
|
||||
client.authenticate()
|
||||
|
||||
postgres_config = None
|
||||
if source == "postgres":
|
||||
connection_id = params.get("connection_id")
|
||||
if not connection_id:
|
||||
log.error("connection_id is required for postgres source.")
|
||||
raise ValueError("connection_id is required for postgres source.")
|
||||
|
||||
# Load connection from DB
|
||||
db = SessionLocal()
|
||||
try:
|
||||
conn_config = db.query(ConnectionConfig).filter(ConnectionConfig.id == connection_id).first()
|
||||
if not conn_config:
|
||||
db_log.error(f"Connection {connection_id} not found.")
|
||||
raise ValueError(f"Connection {connection_id} not found.")
|
||||
|
||||
postgres_config = {
|
||||
'dbname': conn_config.database,
|
||||
'user': conn_config.username,
|
||||
'password': conn_config.password,
|
||||
'host': conn_config.host,
|
||||
'port': str(conn_config.port) if conn_config.port else '5432'
|
||||
}
|
||||
db_log.debug(f"Loaded connection config for {conn_config.host}:{conn_config.port}/{conn_config.database}")
|
||||
finally:
|
||||
db.close()
|
||||
|
||||
log.info(f"Starting mapping for dataset {dataset_id} in {env_name}")
|
||||
|
||||
mapper = DatasetMapper()
|
||||
|
||||
try:
|
||||
mapper.run_mapping(
|
||||
superset_client=client,
|
||||
dataset_id=dataset_id,
|
||||
source=source,
|
||||
postgres_config=postgres_config,
|
||||
excel_path=params.get("excel_path"),
|
||||
table_name=params.get("table_name"),
|
||||
table_schema=params.get("table_schema") or "public"
|
||||
)
|
||||
if source == "sqllab":
|
||||
# Use Superset SQL Lab to query database metadata
|
||||
from ..plugins.translate.superset_executor import SupersetSqlLabExecutor
|
||||
executor = SupersetSqlLabExecutor(config_manager, env_name)
|
||||
database_id = params.get("database_id")
|
||||
if not database_id:
|
||||
log.error("database_id is required for sqllab source.")
|
||||
raise ValueError("database_id is required for sqllab source.")
|
||||
executor.resolve_database_id(target_database_id=str(database_id))
|
||||
|
||||
mapper.run_mapping(
|
||||
superset_client=client,
|
||||
dataset_id=dataset_id,
|
||||
source="sqllab",
|
||||
sqllab_executor=executor,
|
||||
database_id=database_id,
|
||||
sql_query=params.get("sql_query"),
|
||||
)
|
||||
else:
|
||||
# Excel source
|
||||
mapper.run_mapping(
|
||||
superset_client=client,
|
||||
dataset_id=dataset_id,
|
||||
source="excel",
|
||||
excel_path=params.get("excel_path") or params.get("file_data")
|
||||
)
|
||||
superset_log.info(f"Mapping completed for dataset {dataset_id}")
|
||||
return {"status": "success", "dataset_id": dataset_id}
|
||||
except Exception as e:
|
||||
|
||||
Reference in New Issue
Block a user