Fix URL encoding and optimize database connection handling

backend/app/queries/router.py

@@ -1,5 +1,5 @@
 import json
-from urllib.parse import quote_plus
+from urllib.parse import quote
 from fastapi import APIRouter, Depends, HTTPException
 
 from app.auth.utils import CurrentUser, get_current_user, require_role
@@ -155,10 +155,12 @@ def run_live_query(
 ):
     sb = get_supabase()
 
-    # URL-encode user + password so special chars (@, :, #, +, etc.) don't break the URI
+    # URL-encode user, password, and db_name so special chars don't break the URI.
+    # Use quote() with safe='' (not quote_plus) because SQLAlchemy decodes with
+    # urllib.parse.unquote, which does NOT convert '+' back to space.
     conn_str = (
-        f"postgresql+psycopg2://{quote_plus(payload.db_user)}:{quote_plus(payload.db_password)}"
-        f"@{payload.db_host}:{payload.db_port}/{payload.db_name}"
+        f"postgresql+psycopg2://{quote(payload.db_user, safe='')}:{quote(payload.db_password, safe='')}"
+        f"@{payload.db_host}:{payload.db_port}/{quote(payload.db_name, safe='')}"
     )
 
     schema_text = get_external_schema(conn_str, ssl_required=payload.ssl_required)

backend/app/queries/sql_executor.py

@@ -113,7 +113,6 @@ def execute_sql_on_external(
     try:
         engine = create_engine(
             connection_string,
-            pool_pre_ping=True,
             connect_args={"connect_timeout": 10, "sslmode": sslmode},
         )
         with engine.connect() as conn:

backend/app/schema_service/service.py

@@ -42,15 +42,17 @@ def get_external_schema(connection_string: str, ssl_required: bool = True) -> st
     try:
         engine = create_engine(
             connection_string,
-            pool_pre_ping=True,
             connect_args={"connect_timeout": 10, "sslmode": sslmode},
         )
-        inspector = inspect(engine)
         parts = []
-        for table_name in inspector.get_table_names(schema="public"):
-            cols = inspector.get_columns(table_name, schema="public")
-            col_defs = [f"  {c['name']} {c['type']}" for c in cols]
-            parts.append(f"Table: {table_name}\nColumns:\n" + "\n".join(col_defs))
+        # Use a single connection for all reflection calls to avoid the overhead
+        # of opening a new connection per method when using inspect(engine).
+        with engine.connect() as conn:
+            inspector = inspect(conn)
+            for table_name in inspector.get_table_names(schema="public"):
+                cols = inspector.get_columns(table_name, schema="public")
+                col_defs = [f"  {c['name']} {c['type']}" for c in cols]
+                parts.append(f"Table: {table_name}\nColumns:\n" + "\n".join(col_defs))
         return "\n\n".join(parts)
     except Exception as exc:
         msg = str(exc)