refactor(backend): move database helpers to infra

Author: BugMaker-Boyan

packages/backend/app/api/v1/datasources.py

@@ -214,7 +214,7 @@ def list_datasource_tables(
         raise HTTPException(status_code=400, detail="Datasource has no connection_string")
     try:
         from sqlalchemy import create_engine, inspect
-        from app.node.core.db_utils import normalize_connection_string
+        from app.infra.db import normalize_connection_string
         engine = create_engine(normalize_connection_string(ds.connection_string))
         inspector = inspect(engine)
         tables = inspector.get_table_names()

packages/backend/app/datasource/services/connection.py

@@ -4,8 +4,8 @@
 
 from sqlalchemy import inspect, text
 
-from app.node.core.db_utils import create_engine, normalize_connection_string
 from app.datasource.services.specs import normalize_datasource_type, validate_database_datasource_type
+from app.infra.db import create_engine, normalize_connection_string
 
 
 def validate_database_connection(

packages/backend/app/datasource/services/preview.py

@@ -15,8 +15,8 @@
 from sqlalchemy import MetaData, Table, func, inspect, select
 
 from app.core.config import settings
-from app.node.core.db_utils import create_engine, normalize_connection_string
 from app.datasource.services.specs import normalize_datasource_type, validate_database_datasource_type, validate_file_type
+from app.infra.db import create_engine, normalize_connection_string
 from app.infra.services.minio import download_bytes
 
 DEFAULT_PREVIEW_PAGE_SIZE = 25

…ckages/backend/app/node/core/db_utils.py packages/backend/app/infra/db.pypackages/backend/app/node/core/db_utils.py renamed to packages/backend/app/infra/db.py packages/backend/app/node/core/db_utils.py renamed to packages/backend/app/infra/db.py

@@ -6,8 +6,6 @@
 
 from sqlalchemy import text
 
-from app.datasource.services.specs import validate_database_datasource_type
-
 
 def normalize_connection_string(connection_string: str) -> str:
     """Use PyMySQL driver for mysql:// URLs so MySQL works without mysqlclient (MySQLdb)."""
@@ -24,10 +22,6 @@ def create_engine(connection_string: str):
     return create_engine(url)
 
 
-def validate_datasource_type(datasource_type: str | None) -> None:
-    validate_database_datasource_type(datasource_type)
-
-
 def validate_table_name(table: str) -> None:
     if not re.fullmatch(r"[A-Za-z_][A-Za-z0-9_]*", table):
         raise ValueError("Invalid table name")

packages/backend/app/node/core/__init__.py

@@ -1,21 +1,7 @@
 """Core abstractions and shared helpers for workflow nodes."""
 
 from app.node.core.base import BaseNode
-from app.node.core.db_utils import (
-    create_engine,
-    fetch_rows,
-    json_safe_row,
-    normalize_connection_string,
-    validate_datasource_type,
-    validate_table_name,
-)
 
 __all__ = [
     "BaseNode",
-    "create_engine",
-    "fetch_rows",
-    "json_safe_row",
-    "normalize_connection_string",
-    "validate_datasource_type",
-    "validate_table_name",
 ]

packages/backend/app/node/data/sql_execute.py

@@ -4,9 +4,10 @@
 
 from sqlalchemy.orm import Session
 
+from app.datasource.services.specs import validate_database_datasource_type
+from app.infra.db import create_engine, fetch_rows
 from app.repositories import DataSourceRepository
 from app.node.core.base import BaseNode
-from app.node.core.db_utils import create_engine, fetch_rows, validate_datasource_type
 from app.workflow.services.datasets import build_dataset_ref, materialize_sql_query_to_sandbox_result
 from deepeye.workflows.models import Node, Port
 from deepeye.workflows.registry import NodeSpec
@@ -36,7 +37,7 @@ def execute(self, node: Node, inputs: dict[str, Any], context: object) -> dict[s
 
         connection_string = ds.connection_string
         datasource_type = getattr(ds, "type", None)
-        validate_datasource_type(datasource_type)
+        validate_database_datasource_type(datasource_type)
         if not connection_string:
             raise ValueError("database datasource is missing connection_string")
 

packages/backend/app/tasks/agent_datasources.py

@@ -144,7 +144,7 @@ def _get_single_datasource_schema(
         if not connection_string:
             return []
         try:
-            from app.node.core.db_utils import json_safe_row, normalize_connection_string
+            from app.infra.db import json_safe_row, normalize_connection_string
 
             data_engine = create_engine(normalize_connection_string(connection_string))
             try:

packages/backend/app/test/test_architecture_boundaries.py

@@ -39,6 +39,9 @@
     "app.services.docker_control_client",
     "app.services.minio_service",
 }
+LEGACY_NODE_DB_UTILITY_MODULES = {
+    "app.node.core.db_utils",
+}
 LEGACY_WORKFLOW_SERVICE_MODULES = {
     "app.services.workflow_agent_drafts",
     "app.services.workflow_agent_response",
@@ -119,6 +122,18 @@ def test_datasource_domain_does_not_depend_on_tools_layer() -> None:
     assert violations == []
 
 
+def test_datasource_domain_does_not_depend_on_node_layer() -> None:
+    violations: list[str] = []
+    for path in sorted(DATASOURCE_DIR.rglob("*.py")):
+        if path.name == "__init__.py":
+            continue
+        for module in sorted(_imported_modules(path)):
+            if module == "app.node" or module.startswith("app.node."):
+                violations.append(f"{path.relative_to(APP_DIR)} imports {module}")
+
+    assert violations == []
+
+
 def test_runtime_domain_does_not_depend_on_tools_layer() -> None:
     violations: list[str] = []
     for path in sorted(RUNTIME_DIR.rglob("*.py")):
@@ -222,6 +237,16 @@ def test_moved_deploy_and_infra_services_use_domain_import_paths() -> None:
     assert violations == []
 
 
+def test_moved_database_helpers_use_infra_import_paths() -> None:
+    violations: list[str] = []
+    for path in sorted(APP_DIR.rglob("*.py")):
+        legacy_imports = sorted(_imported_modules(path) & LEGACY_NODE_DB_UTILITY_MODULES)
+        for module in legacy_imports:
+            violations.append(f"{path.relative_to(APP_DIR)} imports {module}")
+
+    assert violations == []
+
+
 def test_moved_datasource_services_use_domain_import_paths() -> None:
     violations: list[str] = []
     for path in sorted(APP_DIR.rglob("*.py")):

packages/backend/app/workflow/services/datasets.py

@@ -9,7 +9,7 @@
 
 from sqlalchemy import text
 
-from app.node.core.db_utils import create_engine, json_safe_row
+from app.infra.db import create_engine, json_safe_row
 from app.repositories import DataSourceRepository
 from app.datasource.services.specs import (
     get_datasource_filename,