[NOTICKET] add db_client for querying

src/api/v1/db_client.py

@@ -458,7 +458,7 @@ async def ingest_database_client(
             db_type=client.db_type,
             credentials=creds,
         ) as engine:
-            total = await db_pipeline_service.run(user_id=user_id, engine=engine)
+            total = await db_pipeline_service.run(user_id=user_id, client_id=client_id, engine=engine)
     except NotImplementedError as e:
         raise HTTPException(status_code=status.HTTP_501_NOT_IMPLEMENTED, detail=str(e))
     except Exception as e:

src/pipeline/db_pipeline/db_pipeline_service.py

@@ -148,7 +148,7 @@ class DbPipelineService:
             engine.dispose()
 
     def _to_document(
-        self, user_id: str, table_name: str, entry: dict, updated_at: str
+        self, user_id: str, client_id: str, table_name: str, entry: dict, updated_at: str
     ) -> LangChainDocument:
         col = entry["col"]
         return LangChainDocument(
@@ -156,6 +156,7 @@ class DbPipelineService:
             metadata={
                 "user_id": user_id,
                 "source_type": "database",
+                "database_client_id": client_id,
                 "updated_at": updated_at,
                 "data": {
                     "table_name": table_name,
@@ -170,6 +171,7 @@ class DbPipelineService:
     async def run(
         self,
         user_id: str,
+        client_id: str,
         engine: Engine,
         exclude_tables: Optional[frozenset[str]] = None,
     ) -> int:
@@ -202,7 +204,7 @@ class DbPipelineService:
         for table_name, columns in schema.items():
             logger.info("profiling table", table=table_name, columns=len(columns))
             entries = await asyncio.to_thread(profile_table, engine, table_name, columns)
-            docs = [self._to_document(user_id, table_name, e, updated_at) for e in entries]
+            docs = [self._to_document(user_id, client_id, table_name, e, updated_at) for e in entries]
             if docs:
                 await vector_store.aadd_documents(docs)
                 total += len(docs)