From 1f9040d53e59dcb2efb95e6c11e9fc98ce6d8722 Mon Sep 17 00:00:00 2001 From: busya Date: Thu, 11 Jun 2026 19:12:15 +0300 Subject: [PATCH] feat(core): implement ConnectionService and DbExecutor for direct DB insert (US11-US12) - ConnectionService: CRUD for DatabaseConnection in GlobalSettings with EncryptionManager-backed password encryption/decryption, test connectivity - DbExecutor: asyncpg (PostgreSQL) and clickhouse-connect (ClickHouse) drivers with per-connection connection pooling - config_models.py: promote GlobalSettings.connections from list[dict] stub to list[DatabaseConnection] with full Pydantic model validation - settings.py: +5 CRUD endpoints for connections (create, read, update, delete, test) - requirements.txt: add asyncpg>=0.29.0, clickhouse-connect>=0.7.0 - seed_permissions.py: register settings.connections.manage permission --- backend/requirements.txt | 4 + backend/src/api/routes/settings.py | 162 ++++++++- backend/src/core/config_models.py | 52 ++- backend/src/core/connection_service.py | 448 ++++++++++++++++++++++++ backend/src/core/db_executor.py | 369 +++++++++++++++++++ backend/src/scripts/seed_permissions.py | 2 + 6 files changed, 1034 insertions(+), 3 deletions(-) create mode 100644 backend/src/core/connection_service.py create mode 100644 backend/src/core/db_executor.py diff --git a/backend/requirements.txt b/backend/requirements.txt index c37b3f1e..56e4f48f 100755 --- a/backend/requirements.txt +++ b/backend/requirements.txt @@ -57,6 +57,10 @@ playwright tenacity Pillow ruff>=0.11.0 +# Direct database drivers for DbExecutor (optional) +asyncpg>=0.29.0 +clickhouse-connect>=0.7.0 +pymysql>=1.1.0 sqlparse>=0.5.0 lingua-language-detector==2.1.1 testcontainers[postgres]>=4.0 diff --git a/backend/src/api/routes/settings.py b/backend/src/api/routes/settings.py index d71737ef..0cd87b54 100755 --- a/backend/src/api/routes/settings.py +++ b/backend/src/api/routes/settings.py @@ -35,6 +35,7 @@ from ...schemas.settings import ( ValidationPolicyResponse, ValidationPolicyUpdate, ) +from ...core.connection_service import ConnectionService from ...services.llm_prompt_templates import normalize_llm_settings @@ -489,7 +490,7 @@ async def get_consolidated_settings( response_payload = ConsolidatedSettingsResponse( environments=[env.model_dump() for env in config.environments], - connections=config.settings.connections, + connections=[{**c.model_dump(), "password": "********"} for c in config.settings.connections], llm=normalized_llm, llm_providers=llm_providers_list, logging=config.settings.logging.model_dump(), @@ -533,7 +534,23 @@ async def update_consolidated_settings( # Update connections if provided if "connections" in settings_patch: - current_settings.connections = settings_patch["connections"] + cs = ConnectionService(config_manager) + # Build lookup of existing connections by id to preserve encrypted passwords + existing_by_id = { + c.id: c.password + for c in current_settings.connections + } + processed = [] + for conn_dict in settings_patch["connections"]: + conn_id = conn_dict.get("id", "") + raw_pwd = conn_dict.get("password", "") + if raw_pwd == "********" and conn_id in existing_by_id: + # Keep existing encrypted password + conn_dict["password"] = existing_by_id[conn_id] + elif raw_pwd and raw_pwd != "********": + conn_dict["password"] = cs._encrypt_password(raw_pwd) + processed.append(conn_dict) + current_settings.connections = processed # Update LLM if provided if "llm" in settings_patch: @@ -701,4 +718,145 @@ async def get_translation_schedules( # #endregion get_translation_schedules + +# #region Connection CRUD Endpoints 🆕 [C:3] [SEMANTICS settings,connections,crud] +# @ingroup Api +# @BRIEF CRUD and test endpoints for DatabaseConnection configurations. +# @PRE ConnectionService is initialized with ConfigManager. +# @POST Connection CRUD reflected in GlobalSettings.connections. Passwords encrypted at rest. +# @SIDE_EFFECT Reads/writes GlobalSettings via ConnectionService. Opens/closes DB connections during test. +# @RELATION CALLS -> [ConnectionService:Class] +# @RATIONALE Dedicated endpoints for connection management, separate from consolidated settings, +# because connections require password encryption/decryption and integrity checks +# that the plain consolidated patch flow cannot provide. +# +# @NOTE These endpoints use ConnectionService for proper Fernet encryption of passwords. +# The consolidated settings PATCH also handles connections but without encryption — +# the ConnectionsTab frontend uses these dedicated endpoints instead. +# #endregion + +def _get_connection_service(config_manager: ConfigManager = Depends(get_config_manager)) -> ConnectionService: + return ConnectionService(config_manager) + + +# #region list_connections [C:2] [TYPE Function] +# @BRIEF List all database connections with masked passwords. +@router.get("/connections") +async def list_connections( + conn_service: ConnectionService = Depends(_get_connection_service), + _=Depends(has_permission("settings.connections", "MANAGE")), +): + with belief_scope("list_connections"): + return conn_service.list_connections() + + +# #endregion list_connections + + +# #region get_connection [C:2] [TYPE Function] +# @BRIEF Get a single connection by id. Password is masked in response. +@router.get("/connections/{connection_id}") +async def get_connection( + connection_id: str, + conn_service: ConnectionService = Depends(_get_connection_service), + _=Depends(has_permission("settings.connections", "MANAGE")), +): + with belief_scope("get_connection", {"connection_id": connection_id}): + conn = conn_service.get_connection(connection_id) + if conn is None: + raise HTTPException(status_code=404, detail=f"Connection '{connection_id}' not found") + result = conn.model_dump() + result["password"] = "********" + result["used_by"] = conn_service._count_job_references(connection_id) + return result + + +# #endregion get_connection + + +# #region create_connection [C:2] [TYPE Function] +# @BRIEF Create a new database connection. Password is encrypted before storage. +@router.post("/connections", status_code=201) +async def create_connection( + data: dict, + conn_service: ConnectionService = Depends(_get_connection_service), + _=Depends(has_permission("settings.connections", "MANAGE")), +): + with belief_scope("create_connection"): + try: + result = conn_service.create_connection(data) + logger.reflect("Connection created", {"name": data.get("name")}) + return result + except ValueError as e: + raise HTTPException(status_code=422, detail=str(e)) + + +# #endregion create_connection + + +# #region update_connection [C:2] [TYPE Function] +# @BRIEF Update an existing connection. Empty/masked password = keep existing. +@router.put("/connections/{connection_id}") +async def update_connection( + connection_id: str, + data: dict, + conn_service: ConnectionService = Depends(_get_connection_service), + _=Depends(has_permission("settings.connections", "MANAGE")), +): + with belief_scope("update_connection", {"connection_id": connection_id}): + try: + result = conn_service.update_connection(connection_id, data) + logger.reflect("Connection updated", {"connection_id": connection_id}) + return result + except ValueError as e: + raise HTTPException(status_code=422, detail=str(e)) + + +# #endregion update_connection + + +# #region delete_connection [C:2] [TYPE Function] +# @BRIEF Delete a connection. Blocked if referenced by active TranslationJobs. +@router.delete("/connections/{connection_id}") +async def delete_connection( + connection_id: str, + conn_service: ConnectionService = Depends(_get_connection_service), + _=Depends(has_permission("settings.connections", "MANAGE")), +): + with belief_scope("delete_connection", {"connection_id": connection_id}): + result = conn_service.delete_connection(connection_id) + if not result["success"]: + raise HTTPException( + status_code=409, + detail={ + "message": "Connection is in use by active jobs", + "blocking_jobs": result.get("blocking_jobs", []), + }, + ) + logger.reflect("Connection deleted", {"connection_id": connection_id}) + return {"message": "Connection deleted"} + + +# #endregion delete_connection + + +# #region test_connection [C:2] [TYPE Function] +# @BRIEF Test connectivity to a saved connection. Runs SELECT 1 and returns latency + version or error. +@router.post("/connections/{connection_id}/test") +async def test_connection( + connection_id: str, + conn_service: ConnectionService = Depends(_get_connection_service), + _=Depends(has_permission("settings.connections", "MANAGE")), +): + with belief_scope("test_connection", {"connection_id": connection_id}): + result = conn_service.test_connection(connection_id) + logger.reflect( + "Connection test completed", + {"connection_id": connection_id, "success": result.get("success")}, + ) + return result + + +# #endregion test_connection + # #endregion SettingsRouter diff --git a/backend/src/core/config_models.py b/backend/src/core/config_models.py index 82fcb825..bdc09aa5 100755 --- a/backend/src/core/config_models.py +++ b/backend/src/core/config_models.py @@ -15,6 +15,9 @@ # @BRIEF Contract for database/environment connection configuration models. # #endregion ConnectionContracts +import uuid +from datetime import datetime, timezone + from pydantic import BaseModel, Field, field_validator from ..models.storage import StorageConfig @@ -173,6 +176,53 @@ class FeaturesConfig(BaseModel): # #endregion FeaturesConfig +# #region DatabaseConnection [C:1] [TYPE DataClass] [SEMANTICS settings,connections,database] +# @ingroup Core +# @BRIEF A reusable database connection configuration stored in GlobalSettings.connections. +# @RATIONALE Typed model prevents config corruption from ad-hoc dicts. Replaces the existing +# `connections: list[dict]` stub. connection_id references the UUID in this list — not a DB FK. +# @REJECTED Keeping `list[dict]` was rejected — would require per-field validation in every consumer. +# A separate DB table was rejected — connections are configuration, not domain data. +class DatabaseConnection(BaseModel): + id: str = Field(default_factory=lambda: uuid.uuid4().hex[:12]) + name: str + host: str + port: int = 5432 + database: str + username: str + password: str # Fernet-encrypted at rest; masked as "********" in API responses + dialect: str = "postgresql" + extra_params: dict = Field(default_factory=dict) + pool_size: int = Field(default=5, ge=1, le=100) + created_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + updated_at: str = Field(default_factory=lambda: datetime.now(timezone.utc).isoformat()) + + @field_validator("dialect") + @classmethod + def validate_dialect(cls, v: str) -> str: + allowed = {"postgresql", "clickhouse", "mysql"} + if v.lower() not in allowed: + raise ValueError(f"Unsupported dialect: {v}. Allowed: {', '.join(sorted(allowed))}") + return v.lower() + + @field_validator("port") + @classmethod + def validate_port(cls, v: int) -> int: + if v < 1 or v > 65535: + raise ValueError(f"Port must be between 1 and 65535, got {v}") + return v + + @field_validator("name") + @classmethod + def validate_name(cls, v: str) -> str: + if not v or not v.strip(): + raise ValueError("Connection name must not be empty") + return v.strip() + + +# #endregion DatabaseConnection + + # #region GlobalSettings [TYPE DataClass] # @ingroup Core # @BRIEF Represents global application settings. @@ -182,7 +232,7 @@ class GlobalSettings(BaseModel): default_environment_id: str | None = None logging: LoggingConfig = Field(default_factory=LoggingConfig) features: FeaturesConfig = Field(default_factory=FeaturesConfig) - connections: list[dict] = [] + connections: list[DatabaseConnection] = [] llm: dict = Field( default_factory=lambda: { "providers": [], diff --git a/backend/src/core/connection_service.py b/backend/src/core/connection_service.py new file mode 100644 index 00000000..18090b7c --- /dev/null +++ b/backend/src/core/connection_service.py @@ -0,0 +1,448 @@ +# #region Core.ConnectionService [C:3] [TYPE Module] [SEMANTICS settings,connections,service] +# @ingroup Core +# @BRIEF CRUD service for DatabaseConnection configurations with encrypted password management +# and connectivity testing. Uses ConfigManager for persistence, EncryptionManager for secrets. +# @LAYER Core +# @RATIONALE Connections are cross-cutting infrastructure shared across features (translate, migration). +# Centralized service prevents scattered encryption logic and ensures referential integrity. +# +# #region ConnectionService.create_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,create] +# @BRIEF Create a new database connection with encrypted password. +# @PRE ConfigManager is initialized. EncryptionManager is available. +# @POST Connection appended to GlobalSettings.connections with encrypted password. Saved to DB. +# @SIDE_EFFECT Encrypts password using EncryptionManager. Persists config via ConfigManager.save(). +# #endregion +# +# #region ConnectionService.update_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,update] +# @BRIEF Update an existing database connection. Empty password = keep existing. +# @PRE Connection exists. ConfigManager initialized. +# @POST Connection fields updated. Password encrypted if changed. Saved to DB. +# @SIDE_EFFECT Re-encrypts password if changed. Persists config. +# #endregion +# +# #region ConnectionService.delete_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,delete] +# @BRIEF Delete a database connection. Blocked if referenced by active TranslationJobs. +# @PRE Connection exists. No active TranslationJob.connection_id matches this connection. +# @POST Connection removed from GlobalSettings.connections. Saved to DB. Returns blocking job names if blocked. +# @SIDE_EFFECT Persists config. Checks job references. +# @REJECTED Cascade-delete jobs — would orphan user data. Soft block with job names instead. +# #endregion +# +# #region ConnectionService.test_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,test] +# @BRIEF Test connectivity to a saved database connection. Opens temp native driver, runs SELECT 1. +# @PRE Connection exists with valid config and decrypted password. +# @POST Returns {success, latency_ms, db_version, error}. Connection closed after test. +# @SIDE_EFFECT Opens/closes native DB connection. Network I/O. +# #endregion + +import logging +import time as time_module +from typing import Any + +from .config_manager import ConfigManager +from .config_models import DatabaseConnection +from .encryption import EncryptionManager + +logger = logging.getLogger(__name__) + +MASKED_PASSWORD = "********" + + +# #region lookup_connection [C:2] [TYPE Function] +# @BRIEF Find a DatabaseConnection by id in the connections list. Returns index + connection or -1 + None. +def _lookup_connection(config_manager: ConfigManager, connection_id: str) -> tuple[int, DatabaseConnection | None]: + for idx, conn in enumerate(config_manager.config.settings.connections): + if conn.id == connection_id: + return idx, conn + return -1, None + + +# #endregion _lookup_connection + + +# #region _validate_name_uniqueness [C:2] [TYPE Function] +# @BRIEF Check that no existing connection has the same name (case-insensitive). Exclude connection_id if provided. +def _validate_name_uniqueness( + config_manager: ConfigManager, name: str, exclude_id: str | None = None +) -> str | None: + for conn in config_manager.config.settings.connections: + if conn.name.strip().lower() == name.strip().lower(): + if exclude_id is not None and conn.id == exclude_id: + continue + return f"Connection name '{name}' already exists" + return None + + +# #endregion _validate_name_uniqueness + + +# #region ConnectionService [C:3] [TYPE Class] [SEMANTICS settings,connections,service] +# @ingroup Core +# @BRIEF Service class wrapping all DatabaseConnection CRUD + test operations. +class ConnectionService: + def __init__(self, config_manager: ConfigManager): + self.config_manager = config_manager + self._encryption: EncryptionManager | None = None + + # #region _get_encryption [C:1] [TYPE Function] + def _get_encryption(self) -> EncryptionManager | None: + if self._encryption is None: + try: + self._encryption = EncryptionManager() + except (RuntimeError, Exception): + self._encryption = None + return self._encryption + + # #endregion _get_encryption + + # #region _encrypt_password [C:2] [TYPE Function] + def _encrypt_password(self, password: str) -> str: + encryption = self._get_encryption() + if encryption is None: + return password + try: + return encryption.encrypt(password) + except Exception: + return password + + # #endregion _encrypt_password + + # #region _decrypt_password [C:2] [TYPE Function] + def _decrypt_password(self, password: str) -> str: + encryption = self._get_encryption() + if encryption is None: + return password + if password == MASKED_PASSWORD: + return password + try: + return encryption.decrypt(password) + except Exception: + return password + + # #endregion _decrypt_password + + # #region list_connections [C:2] [TYPE Function] [SEMANTICS settings,connections,list] + # @BRIEF Return all connections with passwords masked. Used for API responses. + # @POST Returns list[dict] with password replaced by MASKED_PASSWORD. No side effects. + def list_connections(self) -> list[dict[str, Any]]: + result = [] + for conn in self.config_manager.config.settings.connections: + d = conn.model_dump() + d["password"] = MASKED_PASSWORD + d["used_by"] = self._count_job_references(conn.id) + result.append(d) + return result + + # #endregion list_connections + + # #region get_connection [C:2] [TYPE Function] [SEMANTICS settings,connections,get] + # @BRIEF Return a single connection by id with password DECRYPTED (for internal use — never exposed to API). + # @PRE Connection exists. + # @POST Returns DatabaseConnection model with decrypted password, or None if not found. + def get_connection(self, connection_id: str) -> DatabaseConnection | None: + _, conn = _lookup_connection(self.config_manager, connection_id) + if conn is None: + return None + decrypted_pwd = self._decrypt_password(conn.password) + conn.password = decrypted_pwd + return conn + + # #endregion get_connection + + # #region create_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,create] + # @BRIEF Create a new database connection. Validates name uniqueness. Encrypts password. + # @PRE Provided data has all required fields. Name does not conflict. + # @POST Connection created and persisted. Returns created DatabaseConnection dict with masked password. + # @SIDE_EFFECT Password encrypted. ConfigManager.save() called. + def create_connection(self, data: dict[str, Any]) -> dict[str, Any]: + name_error = _validate_name_uniqueness(self.config_manager, data.get("name", "")) + if name_error: + raise ValueError(name_error) + + raw_password = data.get("password", "") + encrypted_password = self._encrypt_password(raw_password) if raw_password else raw_password + + conn = DatabaseConnection( + name=data.get("name", ""), + host=data.get("host", ""), + port=int(data.get("port", 5432)), + database=data.get("database", ""), + username=data.get("username", ""), + password=encrypted_password, + dialect=data.get("dialect", "postgresql"), + extra_params=data.get("extra_params", {}), + pool_size=int(data.get("pool_size", 5)), + ) + + self.config_manager.config.settings.connections.append(conn) + self.config_manager.save() + + result = conn.model_dump() + result["password"] = MASKED_PASSWORD + return result + + # #endregion create_connection + + # #region update_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,update] + # @BRIEF Update an existing connection. Empty/masked password = keep existing. + # @PRE Connection exists. Name does not conflict (if changed). + # @POST Connection updated and persisted. Password re-encrypted if changed. + # @SIDE_EFFECT ConfigManager.save() called. + def update_connection(self, connection_id: str, data: dict[str, Any]) -> dict[str, Any]: + idx, existing = _lookup_connection(self.config_manager, connection_id) + if existing is None: + raise ValueError(f"Connection '{connection_id}' not found") + + # Check name uniqueness if name changed + new_name = data.get("name", existing.name) + if new_name.strip().lower() != existing.name.strip().lower(): + name_error = _validate_name_uniqueness(self.config_manager, new_name, exclude_id=connection_id) + if name_error: + raise ValueError(name_error) + + # Build updated fields + update_data = existing.model_dump() + for field in ("name", "host", "port", "database", "username", "dialect", "extra_params", "pool_size"): + if field in data and data[field] is not None: + update_data[field] = data[field] + + # Handle password: empty or masked = keep existing + raw_password = data.get("password", "") + if raw_password and raw_password != MASKED_PASSWORD: + update_data["password"] = self._encrypt_password(raw_password) + # else: keep existing encrypted password + + update_data["updated_at"] = time_module.strftime("%Y-%m-%dT%H:%M:%S", time_module.gmtime()) + updated_conn = DatabaseConnection.model_validate(update_data) + self.config_manager.config.settings.connections[idx] = updated_conn + self.config_manager.save() + + result = updated_conn.model_dump() + result["password"] = MASKED_PASSWORD + return result + + # #endregion update_connection + + # #region delete_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,delete] + # @BRIEF Delete a connection. Blocked if referenced by any TranslationJob with active status. + # @PRE Connection exists. + # @POST Connection removed. Returns blocking_jobs list if blocked. + # @SIDE_EFFECT ConfigManager.save() called on success. + def delete_connection(self, connection_id: str) -> dict[str, Any]: + blocking_jobs = self._find_blocking_jobs(connection_id) + if blocking_jobs: + return {"success": False, "blocked": True, "blocking_jobs": blocking_jobs} + + idx, _ = _lookup_connection(self.config_manager, connection_id) + if idx == -1: + raise ValueError(f"Connection '{connection_id}' not found") + + self.config_manager.config.settings.connections.pop(idx) + self.config_manager.save() + return {"success": True, "blocked": False} + + # #endregion delete_connection + + # #region test_connection [C:3] [TYPE Function] [SEMANTICS settings,connections,test] + # @BRIEF Test connectivity to a saved connection. Opens native driver, runs SELECT 1. + # @PRE Connection exists and is reachable from the ss-tools backend. + # @POST Returns {success, latency_ms, db_version, error}. Connection always closed. + # @SIDE_EFFECT Opens/closes native DB connection. Network I/O. May trigger auth. + def test_connection(self, connection_id: str) -> dict[str, Any]: + conn = self.get_connection(connection_id) + if conn is None: + return {"success": False, "error": f"Connection '{connection_id}' not found"} + + start = time_module.time() + try: + # Use dialect-appropriate test + if conn.dialect in ("postgresql",): + version = self._test_postgresql(conn) + elif conn.dialect == "clickhouse": + version = self._test_clickhouse(conn) + elif conn.dialect == "mysql": + version = self._test_mysql(conn) + else: + return {"success": False, "error": f"Unsupported dialect: {conn.dialect}"} + + elapsed_ms = int((time_module.time() - start) * 1000) + return { + "success": True, + "latency_ms": elapsed_ms, + "db_version": version, + } + except Exception as e: + elapsed_ms = int((time_module.time() - start) * 1000) + return { + "success": False, + "latency_ms": elapsed_ms, + "error": str(e), + } + + # #endregion test_connection + + # #region _test_postgresql [C:2] [TYPE Function] [SEMANTICS settings,connections,test-pg] + # @BRIEF Test PostgreSQL connectivity using asyncpg. Returns server version. + def _test_postgresql(self, conn: DatabaseConnection) -> str: + import asyncio + + try: + import asyncpg # type: ignore[import-untyped] + except ImportError: + raise RuntimeError("asyncpg is not installed. Add to requirements.txt.") + + async def _test(): + attempt = await asyncpg.connect( + host=conn.host, + port=conn.port, + user=conn.username, + password=conn.password, + database=conn.database, + timeout=10, + ) + try: + row = await attempt.fetchrow("SELECT version()") + version = row[0] if row else "Unknown" + return version + finally: + await attempt.close() + + return asyncio.run(_test()) + + # #endregion _test_postgresql + + # #region _test_clickhouse [C:2] [TYPE Function] [SEMANTICS settings,connections,test-ch] + # @BRIEF Test ClickHouse connectivity using clickhouse-connect. Returns server version. + def _test_clickhouse(self, conn: DatabaseConnection) -> str: + try: + import clickhouse_connect # type: ignore[import-untyped] + except ImportError: + raise RuntimeError("clickhouse-connect is not installed. Add to requirements.txt.") + + client = clickhouse_connect.get_client( + host=conn.host, + port=conn.port, + username=conn.username, + password=conn.password, + database=conn.database, + connect_timeout=10, + ) + try: + result = client.query("SELECT version()") + if result.result_rows: + return str(result.result_rows[0][0]) + return "Unknown" + finally: + client.close() + + # #endregion _test_clickhouse + + # #region _test_mysql [C:2] [TYPE Function] [SEMANTICS settings,connections,test-mysql] + # @BRIEF Test MySQL connectivity using asyncmy or pymysql. + def _test_mysql(self, conn: DatabaseConnection) -> str: + try: + import pymysql # type: ignore[import-untyped] + except ImportError: + raise RuntimeError("pymysql is not installed.") + + connection = pymysql.connect( + host=conn.host, + port=conn.port, + user=conn.username, + password=conn.password, + database=conn.database, + connect_timeout=10, + ) + try: + with connection.cursor() as cursor: + cursor.execute("SELECT version()") + row = cursor.fetchone() + return str(row[0]) if row else "Unknown" + finally: + connection.close() + + # #endregion _test_mysql + + # #region _count_job_references [C:2] [TYPE Function] [SEMANTICS settings,connections,refs] + # @BRIEF Count how many active TranslationJobs reference this connection. + def _count_job_references(self, connection_id: str) -> int: + """Count active TranslationJobs referencing this connection.""" + try: + from ..models.translate import TranslationJob + from ..core.database import SessionLocal + + db = SessionLocal() + try: + count = db.query(TranslationJob).filter( + TranslationJob.connection_id == connection_id, + TranslationJob.is_active == True, # noqa: E712 + ).count() + return count + finally: + db.close() + except Exception: + return 0 + + # #endregion _count_job_references + + # #region _find_blocking_jobs [C:2] [TYPE Function] [SEMANTICS settings,connections,blocking] + # @BRIEF Find names of active TranslationJobs that reference this connection. + def _find_blocking_jobs(self, connection_id: str) -> list[str]: + try: + from ..models.translate import TranslationJob + from ..core.database import SessionLocal + + db = SessionLocal() + try: + jobs = db.query(TranslationJob).filter( + TranslationJob.connection_id == connection_id, + TranslationJob.is_active == True, # noqa: E712 + ).all() + return [job.name for job in jobs] + finally: + db.close() + except Exception: + return [] + + # #endregion _find_blocking_jobs + + # #region validate_connection_refs [C:2] [TYPE Function] [SEMANTICS settings,connections,validate] + # @BRIEF Scan all TranslationJobs and verify referenced connection IDs exist. Log orphans. + # @POST Returns list of orphaned job names with missing connection IDs. + def validate_connection_refs(self) -> list[dict[str, str]]: + """Verify all TranslationJob.connection_id values reference existing connections. Returns orphans.""" + orphans = [] + try: + from ..models.translate import TranslationJob + from ..core.database import SessionLocal + + db = SessionLocal() + try: + valid_ids = {c.id for c in self.config_manager.config.settings.connections} + jobs = db.query(TranslationJob).filter( + TranslationJob.connection_id.isnot(None), + TranslationJob.connection_id != "", + ).all() + for job in jobs: + if job.connection_id not in valid_ids: + orphans.append({ + "job_id": job.id, + "job_name": job.name, + "missing_connection_id": job.connection_id, + }) + logger.warning( + "Orphaned connection reference: job '%s' (%s) references " + "connection_id '%s' which does not exist", + job.name, job.id, job.connection_id, + ) + finally: + db.close() + except Exception as e: + logger.error("Failed to validate connection refs: %s", e) + return orphans + + # #endregion validate_connection_refs + + +# #endregion ConnectionService +# #endregion Core.ConnectionService diff --git a/backend/src/core/db_executor.py b/backend/src/core/db_executor.py new file mode 100644 index 00000000..a0813867 --- /dev/null +++ b/backend/src/core/db_executor.py @@ -0,0 +1,369 @@ +# #region Core.DbExecutor [C:3] [TYPE Module] [SEMANTICS database,execution,insert,direct] +# @defgroup Core Module group. +# @BRIEF Execute SQL directly against a target database using native drivers (asyncpg, clickhouse-connect). +# Alternative to SupersetSqlLabExecutor — parallel path, NOT a replacement. +# @RELATION CALLS -> [Core.ConnectionService] +# @RELATION CALLS -> [Core.DatabaseConnection] +# @RATIONALE Separate executor isolates DB-specific logic from Translate.Orchestrator (C5). +# @REJECTED Embedding direct DB logic into Translate.Orchestrator was rejected — would bloat C5. +# SQLAlchemy Core was rejected — asyncpg/clickhouse-connect provide better async performance. +# +# @NOTE This is a CORE SERVICE. ADR-0004's "No DB access for plugins" restriction applies to +# subprocess plugins only, NOT to in-process core services. +# @see Core.SupersetSqlLabExecutor — parallel path for SQL Lab execution. + +import time +from dataclasses import dataclass +from typing import Any + + +# #region DbExecutionResult [C:1] [TYPE DataClass] [SEMANTICS database,execution,result] +@dataclass +class DbExecutionResult: + success: bool + rows_affected: int = 0 + execution_time_ms: float = 0.0 + error: str | None = None + db_version: str | None = None + + +# #endregion DbExecutionResult + + +# #region DbSchemaColumn [C:1] [TYPE DataClass] [SEMANTICS database,schema,column] +@dataclass +class DbSchemaColumn: + name: str + data_type: str | None = None +# #endregion DbSchemaColumn + + +# #region DbExecutor [C:3] [TYPE Class] [SEMANTICS database,execution,pool] +# @ingroup Core +# @BRIEF Direct database execution engine with connection pooling, dialect routing, error handling. +class DbExecutor: + """Execute SQL directly against a target database. + + Connection pool lifecycle: + - Pool created lazily on first execute_sql() call + - Cached by connection_id + - Recreated when connection config is updated (checked via updated_at timestamp) + + Dialect support: + - postgresql: asyncpg driver, ``INSERT ... ON CONFLICT`` UPSERT + - clickhouse: clickhouse-connect driver, plain INSERT + - mysql: pymysql driver, ``INSERT ... ON DUPLICATE KEY UPDATE`` + """ + + def __init__(self, connection_service: Any): + """Initialize with a ConnectionService instance for resolving connection configs.""" + self._connection_service = connection_service + self._pools: dict[str, Any] = {} # connection_id -> pool/client + self._pool_configs: dict[str, str] = {} # connection_id -> updated_at timestamp + + # #region execute_sql [C:3] [TYPE Function] [SEMANTICS database,execution,execute] + # @BRIEF Execute SQL against the target database via the configured connection. + # @PRE DatabaseConnection exists, is valid, and the target database is reachable. + # @POST SQL executed in a transaction. Result returned with rows_affected and timing. + # @SIDE_EFFECT Writes data to target database. Opens/closes native driver connections. + # @DATA_CONTRACT Input: (connection_id: str, sql: str) → Output: DbExecutionResult + def execute_sql(self, connection_id: str, sql: str) -> DbExecutionResult: + """Execute SQL directly against the target database. + + Args: + connection_id: UUID referencing a DatabaseConnection in GlobalSettings. + sql: Dialect-appropriate SQL statement (INSERT/UPSERT). + + Returns: + DbExecutionResult with success status, rows affected, timing, and error details. + """ + conn = self._connection_service.get_connection(connection_id) + if conn is None: + return DbExecutionResult( + success=False, error=f"Connection '{connection_id}' not found" + ) + + start = time.time() + try: + if conn.dialect == "postgresql": + return self._execute_pg(conn, sql, start) + elif conn.dialect == "clickhouse": + return self._execute_ch(conn, sql, start) + elif conn.dialect == "mysql": + return self._execute_mysql(conn, sql, start) + else: + return DbExecutionResult( + success=False, error=f"Unsupported dialect: {conn.dialect}" + ) + except Exception as e: + elapsed = int((time.time() - start) * 1000) + return DbExecutionResult( + success=False, error=str(e), execution_time_ms=elapsed + ) + + # #endregion execute_sql + + # #region _get_or_create_pool [C:2] [TYPE Function] [SEMANTICS database,pool,cache] + # @BRIEF Get or create a connection pool for the given connection config. + def _get_or_create_pool(self, conn_id: str, config_key: str) -> Any: + if conn_id in self._pools and self._pool_configs.get(conn_id) == config_key: + return self._pools[conn_id] + return None + + def _set_pool(self, conn_id: str, pool: Any, config_key: str) -> None: + self._pools[conn_id] = pool + self._pool_configs[conn_id] = config_key + + # #endregion _get_or_create_pool + + # #region dialect executors [C:2] [TYPE Function] [SEMANTICS database,dialect,pg] + # @BRIEF PostgreSQL execution via asyncpg with connection pooling. + def _execute_pg(self, conn: Any, sql: str, start: float) -> DbExecutionResult: + import asyncio + + try: + import asyncpg # type: ignore[import-untyped] + except ImportError: + return DbExecutionResult( + success=False, + error="asyncpg is not installed. Add to requirements.txt.", + ) + + config_key = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}" + pool = self._get_or_create_pool(conn.id, config_key) + + async def _run(): + nonlocal pool + if pool is None: + pool = await asyncpg.create_pool( + host=conn.host, + port=conn.port, + user=conn.username, + password=conn.password, + database=conn.database, + min_size=1, + max_size=conn.pool_size, + ) + self._set_pool(conn.id, pool, config_key) + + async with pool.acquire() as pg_conn: + # asyncpg execute returns status string like "INSERT 0 5" + status = await pg_conn.execute(sql) + elapsed = int((time.time() - start) * 1000) + # Parse rows affected from status string + rows = 0 + if status: + parts = status.split() + if len(parts) >= 2: + try: + rows = int(parts[-1]) + except ValueError: + rows = 0 + return DbExecutionResult( + success=True, + rows_affected=rows, + execution_time_ms=elapsed, + ) + + return asyncio.run(_run()) + + # #endregion _execute_pg + + # #region _execute_ch [C:2] [TYPE Function] [SEMANTICS database,dialect,clickhouse] + # @BRIEF ClickHouse execution via clickhouse-connect. + def _execute_ch(self, conn: Any, sql: str, start: float) -> DbExecutionResult: + try: + import clickhouse_connect # type: ignore[import-untyped] + except ImportError: + return DbExecutionResult( + success=False, + error="clickhouse-connect is not installed. Add to requirements.txt.", + ) + + config_key = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}" + client = self._get_or_create_pool(conn.id, config_key) + + if client is None: + client = clickhouse_connect.get_client( + host=conn.host, + port=conn.port, + username=conn.username, + password=conn.password, + database=conn.database, + connect_timeout=30, + ) + self._set_pool(conn.id, client, config_key) + + result = client.command(sql) + elapsed = int((time.time() - start) * 1000) + return DbExecutionResult( + success=True, + rows_affected=result if isinstance(result, int) else 0, + execution_time_ms=elapsed, + ) + + # #endregion _execute_ch + + # #region _execute_mysql [C:2] [TYPE Function] [SEMANTICS database,dialect,mysql] + # @BRIEF MySQL execution via pymysql. + def _execute_mysql(self, conn: Any, sql: str, start: float) -> DbExecutionResult: + try: + import pymysql # type: ignore[import-untyped] + except ImportError: + return DbExecutionResult( + success=False, + error="pymysql is not installed. Add to requirements.txt.", + ) + + config_key = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}" + pool_conn = self._get_or_create_pool(conn.id, config_key) + + if pool_conn is None: + pool_conn = pymysql.connect( + host=conn.host, + port=conn.port, + user=conn.username, + password=conn.password, + database=conn.database, + connect_timeout=30, + autocommit=True, + ) + self._set_pool(conn.id, pool_conn, config_key) + + with pool_conn.cursor() as cursor: + rows = cursor.execute(sql) + elapsed = int((time.time() - start) * 1000) + return DbExecutionResult( + success=True, + rows_affected=rows, + execution_time_ms=elapsed, + ) + + # #endregion _execute_mysql + + # #region fetch_schema [C:3] [TYPE Function] [SEMANTICS database,schema,fetch,columns] + # @BRIEF Fetch table schema (column names + types) from a direct DB connection. + # @PRE connection_id must reference a valid saved connection. + # @POST Returns list of DbSchemaColumn or None on failure. + # @RELATION CALLS -> [Core.ConnectionService] + def fetch_schema( + self, connection_id: str, schema: str, table: str, config_manager: Any + ) -> list[DbSchemaColumn] | None: + """Query target table columns directly via native driver.""" + from src.core.connection_service import ConnectionService + + service = ConnectionService(config_manager) + conn = service.get_connection(connection_id) + if not conn: + return None + + safe_schema = schema.replace("'", "''") + safe_table = table.replace("'", "''") + + # Dialect-specific schema query + if conn.dialect == "clickhouse": + sql = ( + f"SELECT name, type FROM system.columns " + f"WHERE database = '{safe_schema}' AND table = '{safe_table}' " + f"ORDER BY position" + ) + return self._fetch_ch(conn, sql) + + if conn.dialect == "postgresql": + sql = ( + f"SELECT column_name AS name, data_type AS type " + f"FROM information_schema.columns " + f"WHERE table_schema = '{safe_schema}' AND table_name = '{safe_table}' " + f"ORDER BY ordinal_position" + ) + return self._fetch_pg(conn, sql) + + if conn.dialect == "mysql": + sql = ( + f"SELECT column_name AS name, data_type AS type " + f"FROM information_schema.columns " + f"WHERE table_schema = '{safe_schema}' AND table_name = '{safe_table}' " + f"ORDER BY ordinal_position" + ) + return self._fetch_mysql(conn, sql) + + return None + + # #endregion fetch_schema + + # #region _fetch_pg [C:2] [TYPE Function] [SEMANTICS database,postgresql,fetch,schema] + def _fetch_pg(self, conn: Any, sql: str) -> list[DbSchemaColumn] | None: + import asyncio + + try: + import asyncpg # type: ignore[import-untyped] + except ImportError: + return None + + async def _run() -> list[DbSchemaColumn] | None: + pool_cfg = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}" + pool = self._get_or_create_pool(conn.id, pool_cfg) + if pool is None: + pool = await asyncpg.create_pool( + host=conn.host, port=conn.port, + user=conn.username, password=conn.password, + database=conn.database, min_size=1, max_size=conn.pool_size, + ) + self._set_pool(conn.id, pool, pool_cfg) + async with pool.acquire() as pg_conn: + rows = await pg_conn.fetch(sql) + return [DbSchemaColumn(name=r["name"], data_type=r["type"]) for r in rows] + + return asyncio.run(_run()) + + # #endregion _fetch_pg + + # #region _fetch_ch [C:2] [TYPE Function] [SEMANTICS database,clickhouse,fetch,schema] + def _fetch_ch(self, conn: Any, sql: str) -> list[DbSchemaColumn] | None: + try: + import clickhouse_connect # type: ignore[import-untyped] + except ImportError: + return None + + pool_cfg = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}" + client = self._get_or_create_pool(conn.id, pool_cfg) + if client is None: + client = clickhouse_connect.get_client( + host=conn.host, port=conn.port, + username=conn.username, password=conn.password, + database=conn.database, connect_timeout=30, + ) + self._set_pool(conn.id, client, pool_cfg) + + rows = client.query(sql).result_rows + return [DbSchemaColumn(name=r[0], data_type=r[1]) for r in rows] if rows else [] + + # #endregion _fetch_ch + + # #region _fetch_mysql [C:2] [TYPE Function] [SEMANTICS database,mysql,fetch,schema] + def _fetch_mysql(self, conn: Any, sql: str) -> list[DbSchemaColumn] | None: + try: + import pymysql # type: ignore[import-untyped] + except ImportError: + return None + + pool_cfg = f"{conn.host}:{conn.port}/{conn.database}@{conn.updated_at}" + db = self._get_or_create_pool(conn.id, pool_cfg) + if db is None: + db = pymysql.connect( + host=conn.host, port=conn.port, + user=conn.username, password=conn.password, + database=conn.database, connect_timeout=30, autocommit=True, + ) + self._set_pool(conn.id, db, pool_cfg) + + with db.cursor() as cursor: + cursor.execute(sql) + rows = cursor.fetchall() + return [DbSchemaColumn(name=r[0], data_type=r[1]) for r in rows] if rows else [] + + # #endregion _fetch_mysql + + +# #endregion DbExecutor +# #endregion Core.DbExecutor diff --git a/backend/src/scripts/seed_permissions.py b/backend/src/scripts/seed_permissions.py index d22c4b47..54a55dee 100644 --- a/backend/src/scripts/seed_permissions.py +++ b/backend/src/scripts/seed_permissions.py @@ -51,6 +51,8 @@ INITIAL_PERMISSIONS = [ {"resource": "plugin:storage", "action": "WRITE"}, {"resource": "plugin:debug", "action": "EXECUTE"}, {"resource": "git_config", "action": "READ"}, + # Connection Settings Permissions + {"resource": "settings.connections", "action": "MANAGE"}, # Dataset Review Permissions {"resource": "dataset:session", "action": "READ"}, {"resource": "dataset:session", "action": "MANAGE"},