[NOTICKET][DB] pisahin db credential ke folder model. add ingestion endpoint at db_client to use db pipeline. add router db_client di main.

main.py

@@ -10,6 +10,7 @@ from src.api.v1.chat import router as chat_router
 from src.api.v1.room import router as room_router
 from src.api.v1.users import router as users_router
 from src.api.v1.knowledge import router as knowledge_router
+from src.api.v1.db_client import router as db_client_router
 from src.db.postgres.init_db import init_db
 import uvicorn
 
@@ -35,6 +36,7 @@ app.include_router(document_router)
 app.include_router(knowledge_router)
 app.include_router(room_router)
 app.include_router(chat_router)
+app.include_router(db_client_router)
 
 
 @app.on_event("startup")

src/api/v1/db_client.py

@@ -1,5 +1,362 @@
-from typing import Literal, Dict
+"""API endpoints for user-registered database connections.
 
+Credential schemas (DbType, PostgresCredentials, etc.) live in
+`src/models/credentials.py` — they are imported below (with noqa: F401) so
+FastAPI/Swagger picks them up for OpenAPI schema generation even though they
+are not referenced by name in this file.
+"""
 
-dbtypes: Literal["postgresql", "mysql", "sqlite"] = Literal["postgresql", "mysql", "sqlite"]
-creds: Dict[str, str]
+from typing import Any, Dict, List, Literal, Optional
+from datetime import datetime
+
+from fastapi import APIRouter, Depends, HTTPException, Query, Request, status
+from pydantic import BaseModel, Field
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.database_client.database_client_service import database_client_service
+from src.db.postgres.connection import get_db
+from src.middlewares.logging import get_logger, log_execution
+from src.middlewares.rate_limit import limiter
+from src.models.credentials import (  # noqa: F401 — re-exported for Swagger schema discovery
+    BigQueryCredentials,
+    CredentialSchemas,
+    DbType,
+    MysqlCredentials,
+    PostgresCredentials,
+    SnowflakeCredentials,
+    SqlServerCredentials,
+    SupabaseCredentials,
+)
+from src.pipeline.db_pipeline import db_pipeline_service
+from src.utils.db_credential_encryption import decrypt_credentials_dict
+
+logger = get_logger("database_client_api")
+
+router = APIRouter(prefix="/api/v1", tags=["Database Clients"])
+
+
+# ---------------------------------------------------------------------------
+# Request / Response schemas
+# ---------------------------------------------------------------------------
+
+
+class DatabaseClientCreate(BaseModel):
+    """
+    Payload to register a new external database connection.
+
+    The `credentials` object shape depends on `db_type`:
+
+    | db_type     | Required fields                                          |
+    |-------------|----------------------------------------------------------|
+    | postgres    | host, port, database, username, password, ssl_mode       |
+    | mysql       | host, port, database, username, password, ssl            |
+    | sqlserver   | host, port, database, username, password, driver?        |
+    | supabase    | host, port, database, username, password, ssl_mode       |
+    | bigquery    | project_id, dataset_id, location?, service_account_json  |
+    | snowflake   | account, warehouse, database, schema?, username, password, role? |
+
+    Sensitive fields (`password`, `service_account_json`) are encrypted
+    at rest using Fernet symmetric encryption.
+    """
+
+    name: str = Field(..., description="Display name for this connection.", examples=["Production DB"])
+    db_type: DbType = Field(..., description="Type of the database engine.", examples=["postgres"])
+    credentials: Dict[str, Any] = Field(
+        ...,
+        description="Connection credentials. Shape depends on db_type. See schema descriptions above.",
+        examples=[
+            {
+                "host": "db.example.com",
+                "port": 5432,
+                "database": "mydb",
+                "username": "admin",
+                "password": "s3cr3t!",
+                "ssl_mode": "require",
+            }
+        ],
+    )
+
+
+class DatabaseClientUpdate(BaseModel):
+    """
+    Payload to update an existing database connection.
+
+    All fields are optional — only provided fields will be updated.
+    If `credentials` is provided, it replaces the entire credentials object
+    and sensitive fields are re-encrypted.
+    """
+
+    name: Optional[str] = Field(None, description="New display name for this connection.", examples=["Staging DB"])
+    credentials: Optional[Dict[str, Any]] = Field(
+        None,
+        description="Updated credentials object. Replaces existing credentials entirely if provided.",
+        examples=[{"host": "new-host.example.com", "port": 5432, "database": "mydb", "username": "admin", "password": "n3wP@ss!", "ssl_mode": "require"}],
+    )
+    status: Optional[Literal["active", "inactive"]] = Field(
+        None,
+        description="Set to 'inactive' to soft-disable the connection without deleting it.",
+        examples=["inactive"],
+    )
+
+
+class DatabaseClientResponse(BaseModel):
+    """
+    Database connection record returned by the API.
+
+    Credentials are **never** included in the response for security reasons.
+    """
+
+    id: str = Field(..., description="Unique identifier of the database connection.")
+    user_id: str = Field(..., description="ID of the user who owns this connection.")
+    name: str = Field(..., description="Display name of the connection.")
+    db_type: str = Field(..., description="Database engine type.")
+    status: str = Field(..., description="Connection status: 'active' or 'inactive'.")
+    created_at: datetime = Field(..., description="Timestamp when the connection was registered.")
+    updated_at: Optional[datetime] = Field(None, description="Timestamp of the last update, if any.")
+
+    model_config = {"from_attributes": True}
+
+
+# ---------------------------------------------------------------------------
+# Endpoints
+# ---------------------------------------------------------------------------
+
+
+@router.post(
+    "/database-clients",
+    response_model=DatabaseClientResponse,
+    status_code=status.HTTP_201_CREATED,
+    summary="Register a new database connection",
+    response_description="The newly created database connection record (credentials excluded).",
+    responses={
+        201: {"description": "Connection registered successfully."},
+        422: {"description": "Validation error — check the credentials shape for the given db_type."},
+        500: {"description": "Internal server error."},
+    },
+)
+@limiter.limit("10/minute")
+@log_execution(logger)
+async def create_database_client(
+    request: Request,
+    payload: DatabaseClientCreate,
+    user_id: str = Query(..., description="ID of the user registering the connection."),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Register a new external database connection for a user.
+
+    The `credentials` object must match the shape for the chosen `db_type`
+    (see **CredentialSchemas** in the schema section below for exact fields).
+    Sensitive fields (`password`, `service_account_json`) are encrypted
+    before being persisted — they are never returned in any response.
+    """
+    try:
+        client = await database_client_service.create(
+            db=db,
+            user_id=user_id,
+            name=payload.name,
+            db_type=payload.db_type,
+            credentials=payload.credentials,
+        )
+        return DatabaseClientResponse.model_validate(client)
+    except Exception as e:
+        logger.error(f"Failed to create database client for user {user_id}", error=str(e))
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Failed to create database client: {str(e)}",
+        )
+
+
+@router.get(
+    "/database-clients/{user_id}",
+    response_model=List[DatabaseClientResponse],
+    summary="List all database connections for a user",
+    response_description="List of database connections (credentials excluded).",
+    responses={
+        200: {"description": "Returns an empty list if the user has no connections."},
+    },
+)
+@log_execution(logger)
+async def list_database_clients(
+    user_id: str,
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Return all database connections registered by the specified user,
+    ordered by creation date (newest first).
+
+    Credentials are never included in the response.
+    """
+    clients = await database_client_service.get_user_clients(db, user_id)
+    return [DatabaseClientResponse.model_validate(c) for c in clients]
+
+
+@router.get(
+    "/database-clients/{user_id}/{client_id}",
+    response_model=DatabaseClientResponse,
+    summary="Get a single database connection",
+    response_description="Database connection detail (credentials excluded).",
+    responses={
+        404: {"description": "Connection not found."},
+        403: {"description": "Access denied — user_id does not own this connection."},
+    },
+)
+@log_execution(logger)
+async def get_database_client(
+    user_id: str,
+    client_id: str,
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Return the detail of a single database connection.
+
+    Returns **403** if the `user_id` in the path does not match the owner
+    of the requested connection.
+    """
+    client = await database_client_service.get(db, client_id)
+
+    if not client:
+        raise HTTPException(status_code=404, detail="Database client not found")
+
+    if client.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Access denied")
+
+    return DatabaseClientResponse.model_validate(client)
+
+
+@router.put(
+    "/database-clients/{client_id}",
+    response_model=DatabaseClientResponse,
+    summary="Update a database connection",
+    response_description="Updated database connection record (credentials excluded).",
+    responses={
+        404: {"description": "Connection not found."},
+        403: {"description": "Access denied — user_id does not own this connection."},
+    },
+)
+@log_execution(logger)
+async def update_database_client(
+    client_id: str,
+    payload: DatabaseClientUpdate,
+    user_id: str = Query(..., description="ID of the user who owns the connection."),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Update an existing database connection.
+
+    Only fields present in the request body are updated.
+    If `credentials` is provided it **replaces** the entire credentials object
+    and sensitive fields are re-encrypted automatically.
+    """
+    client = await database_client_service.get(db, client_id)
+
+    if not client:
+        raise HTTPException(status_code=404, detail="Database client not found")
+
+    if client.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Access denied")
+
+    updated = await database_client_service.update(
+        db=db,
+        client_id=client_id,
+        name=payload.name,
+        credentials=payload.credentials,
+        status=payload.status,
+    )
+    return DatabaseClientResponse.model_validate(updated)
+
+
+@router.delete(
+    "/database-clients/{client_id}",
+    status_code=status.HTTP_200_OK,
+    summary="Delete a database connection",
+    responses={
+        200: {"description": "Connection deleted successfully."},
+        404: {"description": "Connection not found."},
+        403: {"description": "Access denied — user_id does not own this connection."},
+    },
+)
+@log_execution(logger)
+async def delete_database_client(
+    client_id: str,
+    user_id: str = Query(..., description="ID of the user who owns the connection."),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Permanently delete a database connection.
+
+    This action is irreversible. The stored credentials are also removed.
+    """
+    client = await database_client_service.get(db, client_id)
+
+    if not client:
+        raise HTTPException(status_code=404, detail="Database client not found")
+
+    if client.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Access denied")
+
+    await database_client_service.delete(db, client_id)
+    return {"status": "success", "message": "Database client deleted successfully"}
+
+
+@router.post(
+    "/database-clients/{client_id}/ingest",
+    status_code=status.HTTP_200_OK,
+    summary="Ingest schema from a registered database into the vector store",
+    response_description="Count of chunks ingested.",
+    responses={
+        200: {"description": "Ingestion completed successfully."},
+        403: {"description": "Access denied — user_id does not own this connection."},
+        404: {"description": "Connection not found."},
+        501: {"description": "The connection's db_type is not yet supported by the pipeline."},
+        500: {"description": "Ingestion failed (connection error, profiling error, etc.)."},
+    },
+)
+@limiter.limit("5/minute")
+@log_execution(logger)
+async def ingest_database_client(
+    request: Request,
+    client_id: str,
+    user_id: str = Query(..., description="ID of the user who owns the connection."),
+    db: AsyncSession = Depends(get_db),
+):
+    """
+    Decrypt the stored credentials, connect to the user's database, introspect
+    its schema, profile each column, embed the descriptions, and store them in
+    the shared PGVector collection tagged with `source_type="database"`.
+
+    Chunks become retrievable via the same retriever used for document chunks.
+    """
+    client = await database_client_service.get(db, client_id)
+
+    if not client:
+        raise HTTPException(status_code=404, detail="Database client not found")
+
+    if client.user_id != user_id:
+        raise HTTPException(status_code=403, detail="Access denied")
+
+    creds = decrypt_credentials_dict(client.credentials)
+
+    try:
+        with db_pipeline_service.engine_scope(
+            db_type=client.db_type,
+            host=creds["host"],
+            port=creds["port"],
+            database=creds["database"],
+            username=creds["username"],
+            password=creds["password"],
+            ssl_mode=creds.get("ssl_mode"),
+        ) as engine:
+            total = await db_pipeline_service.run(user_id=user_id, engine=engine)
+    except NotImplementedError as e:
+        raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail=str(e))
+    except Exception as e:
+        logger.error(
+            f"Ingestion failed for client {client_id}", user_id=user_id, error=str(e)
+        )
+        raise HTTPException(
+            status_code=status.HTTP_500_INTERNAL_SERVER_ERROR,
+            detail=f"Ingestion failed: {e}",
+        )
+
+    return {"status": "success", "client_id": client_id, "chunks_ingested": total}

src/database_client/database_client_service.py

@@ -0,0 +1,118 @@
+"""Service for managing user-registered external database connections."""
+
+import uuid
+from typing import List, Optional
+
+from sqlalchemy import delete, select
+from sqlalchemy.ext.asyncio import AsyncSession
+
+from src.db.postgres.models import DatabaseClient
+from src.middlewares.logging import get_logger
+from src.utils.db_credential_encryption import (
+    decrypt_credentials_dict,
+    encrypt_credentials_dict,
+)
+
+logger = get_logger("database_client_service")
+
+
+class DatabaseClientService:
+    """Service for managing user-registered external database connections."""
+
+    async def create(
+        self,
+        db: AsyncSession,
+        user_id: str,
+        name: str,
+        db_type: str,
+        credentials: dict,
+    ) -> DatabaseClient:
+        """Register a new database client connection.
+
+        Credentials are encrypted before being stored.
+        """
+        client = DatabaseClient(
+            id=str(uuid.uuid4()),
+            user_id=user_id,
+            name=name,
+            db_type=db_type,
+            credentials=encrypt_credentials_dict(credentials),
+            status="active",
+        )
+        db.add(client)
+        await db.commit()
+        await db.refresh(client)
+        logger.info(f"Created database client {client.id} for user {user_id}")
+        return client
+
+    async def get_user_clients(
+        self,
+        db: AsyncSession,
+        user_id: str,
+    ) -> List[DatabaseClient]:
+        """Return all active and inactive database clients for a user."""
+        result = await db.execute(
+            select(DatabaseClient)
+            .where(DatabaseClient.user_id == user_id)
+            .order_by(DatabaseClient.created_at.desc())
+        )
+        return result.scalars().all()
+
+    async def get(
+        self,
+        db: AsyncSession,
+        client_id: str,
+    ) -> Optional[DatabaseClient]:
+        """Return a single database client by its ID."""
+        result = await db.execute(
+            select(DatabaseClient).where(DatabaseClient.id == client_id)
+        )
+        return result.scalars().first()
+
+    async def update(
+        self,
+        db: AsyncSession,
+        client_id: str,
+        name: Optional[str] = None,
+        credentials: Optional[dict] = None,
+        status: Optional[str] = None,
+    ) -> Optional[DatabaseClient]:
+        """Update an existing database client connection.
+
+        Only non-None fields are updated.
+        Credentials are re-encrypted if provided.
+        """
+        client = await self.get(db, client_id)
+        if not client:
+            return None
+
+        if name is not None:
+            client.name = name
+        if credentials is not None:
+            client.credentials = encrypt_credentials_dict(credentials)
+        if status is not None:
+            client.status = status
+
+        await db.commit()
+        await db.refresh(client)
+        logger.info(f"Updated database client {client_id}")
+        return client
+
+    async def delete(
+        self,
+        db: AsyncSession,
+        client_id: str,
+    ) -> bool:
+        """Permanently delete a database client connection."""
+        result = await db.execute(
+            delete(DatabaseClient).where(DatabaseClient.id == client_id)
+        )
+        await db.commit()
+        deleted = result.rowcount > 0
+        if deleted:
+            logger.info(f"Deleted database client {client_id}")
+        return deleted
+
+
+database_client_service = DatabaseClientService()
+ 

src/models/credentials.py

@@ -0,0 +1,164 @@
+"""Pydantic credential schemas for user-registered external databases.
+
+Imported by the `/database-clients` API router (`src/api/v1/db_client.py`) and,
+via `DbType`, by the db pipeline connector (`src/pipeline/db_pipeline/connector.py`).
+
+Sensitive fields (`password`, `service_account_json`) are Fernet-encrypted by
+the database_client service before being stored in the JSONB column; these
+schemas describe the plaintext wire format, not the stored shape.
+"""
+
+from typing import Literal, Optional, Union
+
+from pydantic import BaseModel, Field
+
+# ---------------------------------------------------------------------------
+# Supported DB types
+# ---------------------------------------------------------------------------
+
+DbType = Literal["postgres", "mysql", "sqlserver", "supabase", "bigquery", "snowflake"]
+
+
+# ---------------------------------------------------------------------------
+# Typed credential schemas per DB type
+# ---------------------------------------------------------------------------
+
+
+class PostgresCredentials(BaseModel):
+    """Connection credentials for PostgreSQL."""
+
+    host: str = Field(..., description="Hostname or IP address of the PostgreSQL server.", examples=["db.example.com"])
+    port: int = Field(5432, description="Port number (default: 5432).", examples=[5432])
+    database: str = Field(..., description="Name of the target database.", examples=["mydb"])
+    username: str = Field(..., description="Database username.", examples=["admin"])
+    password: str = Field(..., description="Database password. Will be encrypted at rest.", examples=["s3cr3t!"])
+    ssl_mode: Literal["disable", "require", "verify-ca", "verify-full"] = Field(
+        "require",
+        description="SSL mode for the connection.",
+        examples=["require"],
+    )
+
+
+class MysqlCredentials(BaseModel):
+    """Connection credentials for MySQL."""
+
+    host: str = Field(..., description="Hostname or IP address of the MySQL server.", examples=["db.example.com"])
+    port: int = Field(3306, description="Port number (default: 3306).", examples=[3306])
+    database: str = Field(..., description="Name of the target database.", examples=["mydb"])
+    username: str = Field(..., description="Database username.", examples=["admin"])
+    password: str = Field(..., description="Database password. Will be encrypted at rest.", examples=["s3cr3t!"])
+    ssl: bool = Field(True, description="Enable SSL for the connection.", examples=[True])
+
+
+class SqlServerCredentials(BaseModel):
+    """Connection credentials for Microsoft SQL Server."""
+
+    host: str = Field(..., description="Hostname or IP address of the SQL Server.", examples=["sqlserver.example.com"])
+    port: int = Field(1433, description="Port number (default: 1433).", examples=[1433])
+    database: str = Field(..., description="Name of the target database.", examples=["mydb"])
+    username: str = Field(..., description="Database username.", examples=["sa"])
+    password: str = Field(..., description="Database password. Will be encrypted at rest.", examples=["s3cr3t!"])
+    driver: Optional[str] = Field(
+        None,
+        description="ODBC driver name. Leave empty to use the default driver.",
+        examples=["ODBC Driver 17 for SQL Server"],
+    )
+
+
+class SupabaseCredentials(BaseModel):
+    """Connection credentials for Supabase (PostgreSQL-based).
+
+    Use the connection string details from your Supabase project dashboard
+    under Settings > Database.
+    """
+
+    host: str = Field(
+        ...,
+        description="Supabase database host (e.g. db.<project-ref>.supabase.co, or the pooler host).",
+        examples=["db.xxxx.supabase.co"],
+    )
+    port: int = Field(
+        5432,
+        description="Port number. Use 5432 for direct connection, 6543 for the connection pooler.",
+        examples=[5432],
+    )
+    database: str = Field("postgres", description="Database name (always 'postgres' for Supabase).", examples=["postgres"])
+    username: str = Field(
+        ...,
+        description="Database user. Use 'postgres' for direct connection, or 'postgres.<project-ref>' for the pooler.",
+        examples=["postgres"],
+    )
+    password: str = Field(..., description="Database password (set in Supabase dashboard). Will be encrypted at rest.", examples=["s3cr3t!"])
+    ssl_mode: Literal["require", "verify-ca", "verify-full"] = Field(
+        "require",
+        description="SSL mode. Supabase always requires SSL.",
+        examples=["require"],
+    )
+
+
+class BigQueryCredentials(BaseModel):
+    """Connection credentials for Google BigQuery.
+
+    Requires a GCP Service Account with at least BigQuery Data Viewer
+    and BigQuery Job User roles.
+    """
+
+    project_id: str = Field(..., description="GCP project ID where the BigQuery dataset resides.", examples=["my-gcp-project"])
+    dataset_id: str = Field(..., description="BigQuery dataset name to connect to.", examples=["my_dataset"])
+    location: Optional[str] = Field(
+        "US",
+        description="Dataset location/region (default: US).",
+        examples=["US", "EU", "asia-southeast1"],
+    )
+    service_account_json: str = Field(
+        ...,
+        description=(
+            "Full content of the GCP Service Account key JSON file as a string. "
+            "Will be encrypted at rest."
+        ),
+        examples=['{"type":"service_account","project_id":"my-gcp-project","private_key_id":"..."}'],
+    )
+
+
+class SnowflakeCredentials(BaseModel):
+    """Connection credentials for Snowflake."""
+
+    account: str = Field(
+        ...,
+        description="Snowflake account identifier, including region if applicable (e.g. myaccount.us-east-1).",
+        examples=["myaccount.us-east-1"],
+    )
+    warehouse: str = Field(..., description="Name of the virtual warehouse to use for queries.", examples=["COMPUTE_WH"])
+    database: str = Field(..., description="Name of the target Snowflake database.", examples=["MY_DB"])
+    db_schema: Optional[str] = Field("PUBLIC", alias="schema", description="Schema name (default: PUBLIC).", examples=["PUBLIC"])
+    username: str = Field(..., description="Snowflake username.", examples=["admin"])
+    password: str = Field(..., description="Snowflake password. Will be encrypted at rest.", examples=["s3cr3t!"])
+    role: Optional[str] = Field(None, description="Snowflake role to assume for the session.", examples=["SYSADMIN"])
+
+
+# Union of all credential shapes — reserved for future typed validation on
+# DatabaseClientCreate.credentials (currently Dict[str, Any]). Kept exported
+# so downstream code can reference it without re-declaring.
+CredentialsUnion = Union[
+    PostgresCredentials,
+    MysqlCredentials,
+    SqlServerCredentials,
+    SupabaseCredentials,
+    BigQueryCredentials,
+    SnowflakeCredentials,
+]
+
+
+# Doc-only helper: surfaces per-type credential shapes in the Swagger "Schemas"
+# panel so API consumers can discover the exact field set for each db_type.
+# Not referenced by any endpoint — importing it in db_client.py is enough for
+# FastAPI's OpenAPI generator to pick it up.
+class CredentialSchemas(BaseModel):
+    """Reference schemas for `credentials` per `db_type` (Swagger-only, not used by endpoints)."""
+
+    postgres: PostgresCredentials
+    mysql: MysqlCredentials
+    sqlserver: SqlServerCredentials
+    supabase: SupabaseCredentials
+    bigquery: BigQueryCredentials
+    snowflake: SnowflakeCredentials