ok

app/api/v1/database.py

@@ -6,13 +6,36 @@ from fastapi import APIRouter, Depends, HTTPException, status
 
 from app.api.deps import require_auth
 from app.core.logger import get_logger
-from app.models.schemas import DatabaseQueryRequest, DatabaseQueryResponse
+from app.models.schemas import DatabaseQueryRequest, DatabaseQueryResponse, DatabaseValidateRequest, DatabaseValidateResponse
 from app.services.database_service import DatabaseService
 
 router = APIRouter()
 _logger = get_logger(__name__)
 
 
+@router.post(
+    "/database/validate",
+    response_model=DatabaseValidateResponse,
+    summary="Validate database connection and optionally check table/collection existence",
+)
+async def validate_database(
+    body: DatabaseValidateRequest,
+    token: Annotated[str, Depends(require_auth)],
+    db_service: Annotated[DatabaseService, Depends()] = None,
+) -> DatabaseValidateResponse:
+    if db_service is None:
+        db_service = DatabaseService()
+    _logger.info("Database validate request: %s", body.connection.safe_repr())
+    try:
+        return await db_service.validate_connection(body)
+    except Exception as exc:
+        _logger.error("Unexpected error validating database: %s", exc)
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail={"success": False, "message": f"Internal error: {exc}"},
+        )
+
+
 @router.post(
     "/database/query",
     response_model=DatabaseQueryResponse,

app/core/database/base.py

@@ -19,6 +19,7 @@ class ConnectionConfig:
     database: str
     username: str
     password: str
+    selected_schema: str = "public"
     ssl_enabled: bool = False
     ssl_ca_cert: str | None = None
     ssl_cert: str | None = None
@@ -36,7 +37,7 @@ class ConnectionConfig:
         return (
             f"ConnectionConfig(db_type={self.db_type}, host={self.host}, "
             f"port={self.port}, database={self.database}, "
-            f"username={self.username}, ssl={self.ssl_enabled})"
+            f"selected_schema={self.selected_schema}, username={self.username}, ssl={self.ssl_enabled})"
         )
 
 

app/models/schemas.py

@@ -162,6 +162,7 @@ class DatabaseConnection(BaseModel):
     database: str
     username: str
     password: str = Field(repr=False)
+    selected_schema: str = Field(default="public", description="PostgreSQL schema (ignored for MySQL/MongoDB)")
     ssl: SSLConfig = SSLConfig()
 
     @field_validator("host")
@@ -231,6 +232,7 @@ class DatabaseQueryRequest(BaseModel):
             "database": self.connection.database,
             "username": self.connection.username,
             "password": self.connection.password,
+            "selected_schema": self.connection.selected_schema,
             "ssl_enabled": self.connection.ssl.enabled,
             "ssl_ca_cert": self.connection.ssl.ca_cert,
             "ssl_cert": self.connection.ssl.cert,
@@ -264,6 +266,43 @@ class DatabaseQueryResponse(BaseModel):
     error: Optional[DatabaseQueryError] = None
 
 
+class TableValidationResult(BaseModel):
+    name: str
+    exists: bool
+    error: Optional[str] = None
+
+
+class DatabaseValidateRequest(BaseModel):
+    db_type: Literal["mysql", "postgresql", "mongodb"]
+    connection: DatabaseConnection
+    selected_schema: str = Field(default="public", description="PostgreSQL schema (ignored for MySQL/MongoDB)")
+    table_or_collection_names: List[str] = Field(default_factory=list, description="Optional list of tables/collections to check for existence")
+
+    def to_connection_config(self) -> Dict[str, Any]:
+        return {
+            "db_type": self.db_type,
+            "host": self.connection.host,
+            "port": self.connection.port or {"mysql": 3306, "postgresql": 5432, "mongodb": 27017}[self.db_type],
+            "database": self.connection.database,
+            "username": self.connection.username,
+            "password": self.connection.password,
+            "selected_schema": self.selected_schema,
+            "ssl_enabled": self.connection.ssl.enabled,
+            "ssl_ca_cert": self.connection.ssl.ca_cert,
+            "ssl_cert": self.connection.ssl.cert,
+            "ssl_key": self.connection.ssl.key,
+        }
+
+
+class DatabaseValidateResponse(BaseModel):
+    success: bool
+    time_ms: float
+    message: str
+    connection_details: str
+    error_message: Optional[str] = None
+    tables: List[TableValidationResult] = []
+
+
 class EmbeddingItem(BaseModel):
     success: bool
     time_ms: float

app/services/database_service.py

@@ -10,7 +10,10 @@ from app.models.schemas import (
     DatabaseQueryError,
     DatabaseQueryRequest,
     DatabaseQueryResponse,
+    DatabaseValidateRequest,
+    DatabaseValidateResponse,
     StatementResultSchema,
+    TableValidationResult,
 )
 
 _logger = get_logger(__name__)
@@ -25,6 +28,81 @@ def _root_cause(exc: Exception) -> str:
 
 class DatabaseService:
 
+    async def validate_connection(self, request: DatabaseValidateRequest) -> DatabaseValidateResponse:
+        start_time = time.monotonic()
+        config = ConnectionConfig(**request.to_connection_config())
+        tables = list(request.table_or_collection_names)
+        results: list[TableValidationResult] = []
+
+        try:
+            executor = await pool_manager.get_executor(config)
+        except Exception as exc:
+            elapsed = round((time.monotonic() - start_time) * 1000, 2)
+            _logger.error("Connection failed for %s: %s", config.safe_repr, exc)
+            return DatabaseValidateResponse(
+                success=False,
+                time_ms=elapsed,
+                message="Connection failed",
+                connection_details=config.safe_repr,
+                error_message=_root_cause(exc),
+                tables=[],
+            )
+
+        try:
+            if request.db_type == "mongodb":
+                pool = await executor._get_or_create_pool()
+                db = pool[config.database]
+                await db.command("ping")
+                message = "Connected successfully"
+                if tables:
+                    existing = await db.list_collection_names()
+                    existing_set = set(existing)
+                    for name in tables:
+                        results.append(TableValidationResult(
+                            name=name,
+                            exists=name in existing_set,
+                        ))
+            else:
+                test = await executor.execute(["SELECT 1 AS test"])
+                if not test or not test[0].success:
+                    raise RuntimeError(test[0].error if test else "No response")
+                message = "Connected successfully"
+                if tables:
+                    for name in tables:
+                        if request.db_type == "postgresql":
+                            q = (
+                                f"SELECT 1 FROM information_schema.tables "
+                                f"WHERE table_schema = '{config.selected_schema}' AND table_name = '{name}'"
+                            )
+                        else:
+                            q = (
+                                f"SELECT 1 FROM information_schema.tables "
+                                f"WHERE TABLE_SCHEMA = DATABASE() AND table_name = '{name}'"
+                            )
+                        r = await executor.execute([q])
+                        exists = bool(r and r[0].success and r[0].rows > 0)
+                        results.append(TableValidationResult(name=name, exists=exists))
+        except Exception as exc:
+            elapsed = round((time.monotonic() - start_time) * 1000, 2)
+            _logger.error("Validation failed for %s: %s", config.safe_repr, exc)
+            return DatabaseValidateResponse(
+                success=False,
+                time_ms=elapsed,
+                message="Validation failed",
+                connection_details=config.safe_repr,
+                error_message=_root_cause(exc),
+                tables=results,
+            )
+
+        elapsed = round((time.monotonic() - start_time) * 1000, 2)
+        return DatabaseValidateResponse(
+            success=True,
+            time_ms=elapsed,
+            message=message,
+            connection_details=config.safe_repr,
+            tables=results,
+        )
+
     async def execute_query(self, request: DatabaseQueryRequest) -> DatabaseQueryResponse:
         start_time = time.monotonic()
         config = ConnectionConfig(**request.to_connection_config())