From 30bf2eb61c40e0a5c4797eab3cf0a885fe534775 Mon Sep 17 00:00:00 2001 From: zhoukui <> Date: Sat, 30 Aug 2025 13:46:33 +0800 Subject: [PATCH] datasource info --- src/app/common/rpc.rs | 2 +- src/app/datasource/manager.rs | 24 ++- src/app/datasource/mysql.rs | 17 +- src/app/datasource/postgres.rs | 17 +- src/app/handler/datasource.rs | 344 +++++++++++++++++++++++++++++++++ src/app/handler/mod.rs | 3 +- src/app/server.rs | 32 ++- src/app/startup.rs | 59 ------ 8 files changed, 416 insertions(+), 82 deletions(-) create mode 100644 src/app/handler/datasource.rs diff --git a/src/app/common/rpc.rs b/src/app/common/rpc.rs index 9d54952..e8815f9 100644 --- a/src/app/common/rpc.rs +++ b/src/app/common/rpc.rs @@ -1,5 +1,5 @@ /// 标准RPC响应结构体 -#[derive(Debug, Clone)] +#[derive(Debug, Clone, serde::Serialize)] pub struct RpcResult { /// HTTP状态码，与http::StatusCode一致 pub code: HttpCode, diff --git a/src/app/datasource/manager.rs b/src/app/datasource/manager.rs index f05dc18..26b98ec 100644 --- a/src/app/datasource/manager.rs +++ b/src/app/datasource/manager.rs @@ -50,6 +50,18 @@ impl DatabaseConnection { } } + /// 加载数据库表信息到缓存 + pub async fn load_db_table(&mut self, schema: &str) -> Result<(), Box> { + match self { + DatabaseConnection::Mysql(conn) => { + conn.load_db_table(schema).await.map_err(|e| Box::new(e) as Box) + } + DatabaseConnection::Postgres(conn) => { + conn.load_db_table(schema).await.map_err(|e| Box::new(e) as Box) + } + } + } + /// 查询记录数 pub async fn count(&self, sql: &str, params: Vec) -> Result> { match self { @@ -114,13 +126,21 @@ impl DataSourceManager { let db_connection = match datasource.kind { DataSourceKind::Mysql => { - let conn = DBConn::new(&connection_url).await + let mut conn = DBConn::new(&connection_url, &datasource.name).await .map_err(|e| format!("Failed to connect to MySQL database {}: {}", database, e))?; + // 加载表信息到缓存 + if let Err(e) = conn.load_db_table(database).await { + log::warn!("Failed to load table metadata for MySQL database {}: {}", database, e); + } DatabaseConnection::Mysql(conn) } DataSourceKind::Postgres => { - let conn = PgConn::new(&connection_url).await + let mut conn = PgConn::new(&connection_url, &datasource.name).await .map_err(|e| format!("Failed to connect to PostgreSQL database {}: {}", database, e))?; + // 加载表信息到缓存 + if let Err(e) = conn.load_db_table(database).await { + log::warn!("Failed to load table metadata for PostgreSQL database {}: {}", database, e); + } DatabaseConnection::Postgres(conn) } }; diff --git a/src/app/datasource/mysql.rs b/src/app/datasource/mysql.rs index cfe5b78..5d8ed22 100644 --- a/src/app/datasource/mysql.rs +++ b/src/app/datasource/mysql.rs @@ -5,7 +5,7 @@ use sqlx::{ types::Decimal, }; use std::collections::HashMap; -use crate::app::datasource::metadata::{put_db_meta, put_db_tables, put_table_meta}; +use crate::app::datasource::metadata::{put_db_meta, put_datasource_db_tables, put_datasource_table_meta}; use crate::app::datasource::codec::base64_encode; use crate::app::datasource::{ColumnMeta, TableMeta}; @@ -22,6 +22,8 @@ const MYSQL_SYS_DB: &[&str] = &["information_schema", "mysql", "performance_sche pub struct DBConn { /// MySQL 连接池 pool: MySqlPool, + /// 数据源名称 + datasource_name: String, } impl DBConn { @@ -35,9 +37,12 @@ impl DBConn { /// # 返回值 /// * `Ok(Self)` - 成功创建的数据库连接实例 /// * `Err(sqlx::Error)` - 连接或初始化过程中发生的错误 - pub async fn new(url: &str) -> Result { + pub async fn new(url: &str, datasource_name: &str) -> Result { let pool = MySqlPool::connect(url).await?; - let mut ds = Self { pool }; + let mut ds = Self { + pool, + datasource_name: datasource_name.to_string(), + }; ds.init().await?; Ok(ds) } @@ -113,7 +118,7 @@ impl DBConn { /// # 返回值 /// * `Ok(())` - 成功加载所有表信息 /// * `Err(sqlx::Error)` - 查询或处理过程中发生的错误 - async fn load_db_table(&mut self, schema: &str) -> Result<(), sqlx::Error> { + pub async fn load_db_table(&mut self, schema: &str) -> Result<(), sqlx::Error> { // 构造查询表信息的 SQL 语句，只查询基础表（BASE TABLE） let list_db_table_sql = format!( "SELECT TABLE_NAME, TABLE_COMMENT @@ -165,14 +170,14 @@ impl DBConn { log::info!("mysql.table: {}.{} loaded", schema, &table_name); // 将表元数据存入外部缓存 - put_table_meta(schema, &table_name, table_meta); + put_datasource_table_meta(&self.datasource_name, schema, &table_name, table_meta); // 将表名添加到表名列表中 table_name_list.push(table_name); } // 将当前数据库的表列表存入外部缓存 - put_db_tables(schema.to_string(), table_name_list); + put_datasource_db_tables(&self.datasource_name, schema.to_string(), table_name_list); Ok(()) } diff --git a/src/app/datasource/postgres.rs b/src/app/datasource/postgres.rs index 51c841b..78a1a0f 100644 --- a/src/app/datasource/postgres.rs +++ b/src/app/datasource/postgres.rs @@ -5,7 +5,7 @@ use sqlx::{ types::Decimal, }; use std::collections::HashMap; -use crate::app::datasource::metadata::{put_db_meta, put_db_tables, put_table_meta}; +use crate::app::datasource::metadata::{put_db_meta, put_datasource_db_tables, put_datasource_table_meta}; use crate::app::datasource::codec::base64_encode; use crate::app::datasource::{ColumnMeta, TableMeta}; @@ -22,6 +22,8 @@ const PG_SYS_DB: &[&str] = &["information_schema", "pg_catalog", "pg_toast", "po pub struct PgConn { /// PostgreSQL 连接池 pool: PgPool, + /// 数据源名称 + datasource_name: String, } impl PgConn { @@ -35,9 +37,12 @@ impl PgConn { /// # 返回值 /// * `Ok(Self)` - 成功创建的数据库连接实例 /// * `Err(sqlx::Error)` - 连接或初始化过程中发生的错误 - pub async fn new(url: &str) -> Result { + pub async fn new(url: &str, datasource_name: &str) -> Result { let pool = PgPool::connect(url).await?; - let mut ds = Self { pool }; + let mut ds = Self { + pool, + datasource_name: datasource_name.to_string(), + }; ds.init().await?; Ok(ds) } @@ -108,7 +113,7 @@ impl PgConn { /// # 返回值 /// * `Ok(())` - 成功加载所有表信息 /// * `Err(sqlx::Error)` - 查询或处理过程中发生的错误 - async fn load_db_table(&mut self, schema: &str) -> Result<(), sqlx::Error> { + pub async fn load_db_table(&mut self, schema: &str) -> Result<(), sqlx::Error> { // 构造查询表信息的 SQL 语句，只查询基础表（BASE TABLE） let list_db_table_sql = format!( "SELECT t.table_name, @@ -153,14 +158,14 @@ impl PgConn { log::info!("postgres.table: {}.{} loaded", schema, &table_name); // 将表元数据存入外部缓存 - put_table_meta(schema, &table_name, table_meta); + put_datasource_table_meta(&self.datasource_name, schema, &table_name, table_meta); // 将表名添加到表名列表中 table_name_list.push(table_name); } // 将当前数据库的表列表存入外部缓存 - put_db_tables(schema.to_string(), table_name_list); + put_datasource_db_tables(&self.datasource_name, schema.to_string(), table_name_list); Ok(()) } diff --git a/src/app/handler/datasource.rs b/src/app/handler/datasource.rs new file mode 100644 index 0000000..091157f --- /dev/null +++ b/src/app/handler/datasource.rs @@ -0,0 +1,344 @@ +use std::collections::HashMap; +use log::{info, debug, warn}; +use serde_json::{json, Value}; +use crate::app::common::rpc::{RpcResult, HttpCode}; +use crate::app::datasource::manager::DataSourceManager; +use crate::app::datasource::metadata; + +/// 处理数据源相关查询请求的统一入口 +/// +/// 根据action参数区分不同的查询类型： +/// - "info": 返回数据源信息（数据库和表的树形结构） +/// - "tables": 返回指定数据库下所有表的详细信息 +/// +/// # 参数 +/// * `manager` - 数据源管理器 +/// * `action` - 操作类型（"info" 或 "tables"） +/// * `body_map` - 包含请求参数的HashMap +/// +/// 对于action="info"： +/// - `datasource`: 可选，指定数据源名称，如果不提供则返回所有数据源信息 +/// +/// 对于action="tables"： +/// - `datasource`: 必需，数据源名称 +/// - `database`: 必需，数据库名称 +pub async fn handle_datasource( + manager: &DataSourceManager, + action: &str, + body_map: HashMap, +) -> RpcResult> { + info!("开始处理数据源查询请求，操作类型: {}", action); + debug!("请求参数: {:?}", body_map); + + match action { + "info" => handle_datasource_info(manager, body_map).await, + "tables" => handle_database_tables(manager, body_map).await, + _ => { + let rpc_result = RpcResult { + code: HttpCode::BadRequest, + msg: Some(format!("不支持的操作类型: {}", action)), + data: None, + }; + rpc_result + } + } +} + +/// 处理数据源信息查询请求 +/// +/// 返回指定数据源下的数据库及数据表信息的树形结构 +/// +/// # 参数 +/// * `manager` - 数据源管理器 +/// * `body_map` - 包含请求参数的HashMap，可选参数： +/// - `datasource`: 指定数据源名称，如果不提供则返回所有数据源信息 +/// +/// # 返回值 +/// 返回树形结构的JSON数据： +/// ```json +/// { +/// "datasource_name": { +/// "type": "datasource", +/// "name": "datasource_name", +/// "databases": { +/// "database_name": { +/// "type": "database", +/// "name": "database_name", +/// "tables": { +/// "table_name": { +/// "type": "table", +/// "name": "table_name", +/// "comment": "table_comment" +/// } +/// } +/// } +/// } +/// } +/// } +/// ``` +pub async fn handle_datasource_info( + manager: &DataSourceManager, + body_map: HashMap, +) -> RpcResult> { + info!("开始处理数据源信息查询请求"); + debug!("请求参数: {:?}", body_map); + + let mut rpc_result = RpcResult { + code: HttpCode::Ok, + msg: None, + data: None, + }; + + // 检查是否指定了特定数据源 + let target_datasource = body_map.get("datasource") + .and_then(|v| v.as_str()) + .map(|s| s.to_string()); + + let mut result_data = HashMap::new(); + + // 获取要处理的数据源列表 + let datasource_names = if let Some(ds_name) = target_datasource { + // 验证指定的数据源是否存在 + let all_datasources = manager.get_datasource_names(); + if all_datasources.contains(&ds_name) { + vec![ds_name] + } else { + rpc_result.code = HttpCode::NotFound; + rpc_result.msg = Some(format!("数据源 '{}' 不存在", ds_name)); + return rpc_result; + } + } else { + manager.get_datasource_names() + }; + + info!("处理 {} 个数据源", datasource_names.len()); + + // 遍历每个数据源 + for datasource_name in datasource_names { + debug!("处理数据源: {}", datasource_name); + + let mut datasource_info = HashMap::new(); + datasource_info.insert("type".to_string(), json!("datasource")); + datasource_info.insert("name".to_string(), json!(datasource_name.clone())); + + // 获取数据源下的所有数据库 + let database_names = manager.get_database_names(&datasource_name); + let mut databases_info = HashMap::new(); + + debug!("数据源 {} 包含 {} 个数据库", datasource_name, database_names.len()); + + // 遍历每个数据库 + for database_name in database_names { + debug!("处理数据库: {}.{}", datasource_name, database_name); + + let mut database_info = HashMap::new(); + database_info.insert("type".to_string(), json!("database")); + database_info.insert("name".to_string(), json!(database_name.clone())); + + // 获取数据库下的所有表信息 + let table_info_map = metadata::get_datasource_table_name_list(&datasource_name, &database_name); + let mut tables_info = HashMap::new(); + + debug!("数据库 {}.{} 包含 {} 个表", datasource_name, database_name, table_info_map.len()); + + // 遍历每个表 + for (table_name, table_comment) in table_info_map { + let mut table_info = HashMap::new(); + table_info.insert("type".to_string(), json!("table")); + table_info.insert("name".to_string(), json!(table_name.clone())); + table_info.insert("comment".to_string(), json!(table_comment)); + + tables_info.insert(table_name, json!(table_info)); + } + + database_info.insert("tables".to_string(), json!(tables_info)); + databases_info.insert(database_name, json!(database_info)); + } + + datasource_info.insert("databases".to_string(), json!(databases_info)); + result_data.insert(datasource_name, json!(datasource_info)); + } + + info!("数据源信息查询完成，返回 {} 个数据源的信息", result_data.len()); + + rpc_result.data = Some(result_data); + rpc_result +} + +/// 处理数据库表详情查询请求 +/// +/// 返回指定数据库下所有表的详细信息，包括表结构、字段信息等 +/// +/// # 参数 +/// * `manager` - 数据源管理器 +/// * `body_map` - 包含请求参数的HashMap，必需参数： +/// - `datasource`: 数据源名称 +/// - `database`: 数据库名称 +/// +/// # 返回值 +/// 返回树形结构的JSON数据： +/// ```json +/// { +/// "database_info": { +/// "type": "database", +/// "datasource": "datasource_name", +/// "name": "database_name", +/// "tables": { +/// "table_name": { +/// "type": "table", +/// "name": "table_name", +/// "comment": "table_comment", +/// "columns": { +/// "column_name": { +/// "type": "column", +/// "name": "column_name", +/// "field": "column_name", +/// "type_name": "varchar", +/// "null": "YES", +/// "default": null, +/// "comment": "column_comment", +/// "key": "", +/// "extra": "" +/// } +/// } +/// } +/// } +/// } +/// } +/// ``` +pub async fn handle_database_tables( + manager: &DataSourceManager, + body_map: HashMap, +) -> RpcResult> { + info!("开始处理数据库表详情查询请求"); + debug!("请求参数: {:?}", body_map); + + let mut rpc_result = RpcResult { + code: HttpCode::Ok, + msg: None, + data: None, + }; + + // 获取必需的参数 + let datasource_name = match body_map.get("datasource").and_then(|v| v.as_str()) { + Some(name) => name, + None => { + rpc_result.code = HttpCode::BadRequest; + rpc_result.msg = Some("缺少必需参数: datasource".to_string()); + return rpc_result; + } + }; + + let database_name = match body_map.get("database").and_then(|v| v.as_str()) { + Some(name) => name, + None => { + rpc_result.code = HttpCode::BadRequest; + rpc_result.msg = Some("缺少必需参数: database".to_string()); + return rpc_result; + } + }; + + info!("查询数据库表详情: {}.{}", datasource_name, database_name); + + // 验证数据源是否存在 + let all_datasources = manager.get_datasource_names(); + if !all_datasources.contains(&datasource_name.to_string()) { + rpc_result.code = HttpCode::NotFound; + rpc_result.msg = Some(format!("数据源 '{}' 不存在", datasource_name)); + return rpc_result; + } + + // 验证数据库是否存在 + let databases = manager.get_database_names(datasource_name); + if !databases.contains(&database_name.to_string()) { + rpc_result.code = HttpCode::NotFound; + rpc_result.msg = Some(format!("数据库 '{}.{}' 不存在", datasource_name, database_name)); + return rpc_result; + } + + // 构建数据库信息 + let mut database_info = HashMap::new(); + database_info.insert("type".to_string(), json!("database")); + database_info.insert("datasource".to_string(), json!(datasource_name)); + database_info.insert("name".to_string(), json!(database_name)); + + // 获取数据库下的所有表信息 + let table_info_map = metadata::get_datasource_table_name_list(datasource_name, database_name); + let mut tables_info = HashMap::new(); + + info!("数据库 {}.{} 包含 {} 个表", datasource_name, database_name, table_info_map.len()); + + // 遍历每个表，获取详细信息 + for (table_name, table_comment) in table_info_map { + debug!("处理表: {}.{}.{}", datasource_name, database_name, table_name); + + let mut table_info = HashMap::new(); + table_info.insert("type".to_string(), json!("table")); + table_info.insert("name".to_string(), json!(table_name.clone())); + table_info.insert("comment".to_string(), json!(table_comment)); + + // 获取表的元数据信息 + if let Some(table_meta) = metadata::get_datasource_table(datasource_name, database_name, &table_name) { + let mut columns_info = HashMap::new(); + + debug!("表 {} 包含 {} 个字段", table_name, table_meta.columns.len()); + + // 遍历表的每个字段 + for (_column_name, column) in &table_meta.columns { + let mut column_info = HashMap::new(); + column_info.insert("type".to_string(), json!("column")); + column_info.insert("name".to_string(), json!(column.field.clone())); + column_info.insert("field".to_string(), json!(column.field.clone())); + column_info.insert("type_name".to_string(), json!(column.type_name.clone())); + column_info.insert("null".to_string(), json!(column.null.clone())); + column_info.insert("default".to_string(), + if let Some(ref default) = column.default { + json!(default) + } else { + json!(null) + } + ); + column_info.insert("comment".to_string(), + if let Some(ref comment) = column.comment { + json!(comment) + } else { + json!("") + } + ); + column_info.insert("key".to_string(), + if let Some(ref key) = column.key { + json!(key) + } else { + json!("") + } + ); + column_info.insert("extra".to_string(), + if let Some(ref extra) = column.extra { + json!(extra) + } else { + json!("") + } + ); + + columns_info.insert(column.field.clone(), json!(column_info)); + } + + table_info.insert("columns".to_string(), json!(columns_info)); + } else { + warn!("无法获取表 {}.{}.{} 的元数据信息", datasource_name, database_name, table_name); + table_info.insert("columns".to_string(), json!({})); + } + + tables_info.insert(table_name, json!(table_info)); + } + + database_info.insert("tables".to_string(), json!(tables_info)); + + let mut result_data = HashMap::new(); + result_data.insert("database_info".to_string(), json!(database_info)); + + info!("数据库表详情查询完成，返回 {} 个表的详细信息", tables_info.len()); + + rpc_result.data = Some(result_data); + rpc_result +} \ No newline at end of file diff --git a/src/app/handler/mod.rs b/src/app/handler/mod.rs index 6b75e3d..8310655 100644 --- a/src/app/handler/mod.rs +++ b/src/app/handler/mod.rs @@ -4,4 +4,5 @@ pub mod get; pub mod delete; pub mod head; pub mod post; -pub mod put; \ No newline at end of file +pub mod put; +pub mod datasource; \ No newline at end of file diff --git a/src/app/server.rs b/src/app/server.rs index b7547ff..59875fc 100644 --- a/src/app/server.rs +++ b/src/app/server.rs @@ -4,7 +4,7 @@ use std::sync::Arc; use axum::{ - extract::Path, + extract::{Path, Query}, Json, routing::{get, post}, Router, @@ -20,6 +20,7 @@ use crate::app::{ post::handle_post, put::handle_put, delete::handle_delete, + datasource::handle_datasource, }, }; @@ -54,11 +55,24 @@ pub fn create_router(datasource_manager: Arc) -> Router { })); } }; - Json(json!({ - "code": rpc_result.code as u16, - "data": rpc_result.data, - "msg": rpc_result.msg - })) + Json(serde_json::to_value(rpc_result).unwrap()) + } + }; + + // 数据源相关操作处理器 + let mgr_datasource = datasource_manager.clone(); + let datasource_handler = move |Path((datasource, action)): Path<(String, String)>, Query(query_params): Query>| { + let mgr = mgr_datasource.clone(); + async move { + // 将路径参数和查询参数合并 + let mut request_data: HashMap = query_params + .into_iter() + .map(|(k, v)| (k, serde_json::Value::String(v))) + .collect(); + // 将datasource参数添加到请求数据中 + request_data.insert("datasource".to_string(), serde_json::Value::String(datasource)); + let rpc_result = handle_datasource(&mgr, &action, request_data).await; + Json(serde_json::to_value(rpc_result).unwrap()) } }; @@ -66,6 +80,8 @@ pub fn create_router(datasource_manager: Arc) -> Router { // 动态派发CRUD操作，形如：/get.json、/post.json、/put.json、/delete.json、/head.json // 这里使用 `/:method` 捕获，包括带有 .json 后缀的情形（例如 get.json），在处理器中去掉后缀 .route("/:method", post(curd_handler)) + // 数据源相关操作接口（动态路由） + .route("/:datasource/:action", get(datasource_handler)) .route("/health", get(health_check)) .layer(CorsLayer::permissive()) } @@ -89,7 +105,9 @@ pub async fn start_server( println!("🚀 服务器启动成功！"); println!("📍 监听地址: http://localhost:{}", port); - println!("🔍 API示例: http://localhost:{}/api/v1/ds_mysql/panda_db/user", port); + println!("🔍 API示例: http://localhost:{}/[get|post|put|delete|head].json", port); + println!("📊 数据源信息: http://localhost:{}/ds_mysql/info", port); + println!("📋 数据库表详情: http://localhost:{}/ds_mysql/tables?database=panda_db", port); println!("💚 健康检查: http://localhost:{}/health", port); axum::serve(listener, app).await?; diff --git a/src/app/startup.rs b/src/app/startup.rs index 71f2c8b..24bf9ef 100644 --- a/src/app/startup.rs +++ b/src/app/startup.rs @@ -146,63 +146,4 @@ impl AppStartup { stats.join("\n") } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[tokio::test] - async fn test_initialize_with_invalid_config() { - // 测试无效配置文件 - let result = AppStartup::initialize(Some("non_existent_file.yaml")).await; - assert!(result.is_err()); - - if let Err(StartupError::ConfigError(_)) = result { - // 预期的错误类型 - } else { - panic!("Expected ConfigError"); - } - } - - #[test] - fn test_startup_error_display() { - let config_error = StartupError::ConfigError( - ConfigLoadError::ValidationError("测试错误".to_string()) - ); - assert!(config_error.to_string().contains("配置错误")); - - let datasource_error = StartupError::DataSourceError("连接失败".to_string()); - assert!(datasource_error.to_string().contains("数据源错误")); - - let other_error = StartupError::Other("其他错误".to_string()); - assert!(other_error.to_string().contains("启动错误")); - } - - #[test] - fn test_get_statistics() { - // 这个测试需要一个有效的DataSourceManager实例 - // 由于依赖外部资源，这里只测试函数不会panic - use crate::app::datasource::config::{DataSourcesConfig, DataSourceConfig, DataSourceKind}; - - let config = DataSourcesConfig { - datasource: vec![ - DataSourceConfig { - name: "test_ds".to_string(), - kind: DataSourceKind::Mysql, - username: "root".to_string(), - password: "123456".to_string(), - url: "mysql://localhost:3306".to_string(), - database: vec!["test_db".to_string()], - default: true, - }, - ], - }; - - let manager = DataSourceManager::new(config); - let stats = AppStartup::get_statistics(&manager); - - assert!(stats.contains("数据源总数")); - assert!(stats.contains("数据库总数")); - } } \ No newline at end of file