diff --git a/lang_agent/config/db_config_manager.py b/lang_agent/config/db_config_manager.py index 25624bf..bb5ce7c 100644 --- a/lang_agent/config/db_config_manager.py +++ b/lang_agent/config/db_config_manager.py @@ -17,6 +17,76 @@ class DBConfigManager: cur.execute("DELETE FROM prompt_sets WHERE pipeline_id = %s AND id = %s", (pipeline_id, prompt_set_id)) conn.commit() + def list_prompt_sets(self, pipeline_id: Optional[str] = None) -> List[Dict[str, object]]: + """ + List prompt_set metadata for UI listing. + """ + with psycopg.connect(self.conn_str) as conn: + with conn.cursor(row_factory=dict_row) as cur: + if pipeline_id: + cur.execute( + """ + SELECT id, pipeline_id, name, description, is_active, created_at, updated_at, list + FROM prompt_sets + WHERE pipeline_id = %s + ORDER BY updated_at DESC, created_at DESC + """, + (pipeline_id,), + ) + else: + cur.execute( + """ + SELECT id, pipeline_id, name, description, is_active, created_at, updated_at, list + FROM prompt_sets + ORDER BY updated_at DESC, created_at DESC + """ + ) + rows = cur.fetchall() + + return [ + { + "prompt_set_id": str(row["id"]), + "pipeline_id": row["pipeline_id"], + "name": row["name"], + "description": row["description"] or "", + "is_active": bool(row["is_active"]), + "created_at": row["created_at"].isoformat() if row["created_at"] else None, + "updated_at": row["updated_at"].isoformat() if row["updated_at"] else None, + "tool_keys": self._parse_tool_list(row.get("list")), + } + for row in rows + ] + + def get_prompt_set(self, pipeline_id: str, prompt_set_id: str) -> Optional[Dict[str, object]]: + """ + Return prompt_set metadata by id within a pipeline. + """ + with psycopg.connect(self.conn_str) as conn: + with conn.cursor(row_factory=dict_row) as cur: + cur.execute( + """ + SELECT id, pipeline_id, name, description, is_active, created_at, updated_at, list + FROM prompt_sets + WHERE id = %s AND pipeline_id = %s + """, + (prompt_set_id, pipeline_id), + ) + row = cur.fetchone() + + if row is None: + return None + + return { + "prompt_set_id": str(row["id"]), + "pipeline_id": row["pipeline_id"], + "name": row["name"], + "description": row["description"] or "", + "is_active": bool(row["is_active"]), + "created_at": row["created_at"].isoformat() if row["created_at"] else None, + "updated_at": row["updated_at"].isoformat() if row["updated_at"] else None, + "tool_keys": self._parse_tool_list(row.get("list")), + } + def get_config( self, pipeline_id: str, prompt_set_id: Optional[str] = None ) -> Tuple[Dict[str, str], List[str]]: