feat(schema): 添加 schema snapshot 编排

- 新增 schema snapshot DTO,标准化连接、数据库、driver 和采集时间信息

- 复用现有 schema core 函数采集表、视图、列、索引、外键和触发器

- 添加 SQLite 集成测试覆盖表/视图与元数据快照输出
This commit is contained in:
Illuminated2020
2026-05-12 16:47:40 +08:00
parent 037e082104
commit 78be118c58
3 changed files with 219 additions and 0 deletions
+1
View File
@@ -14,6 +14,7 @@ pub mod query_cancel;
pub mod redis_ops;
pub mod saved_sql;
pub mod schema;
pub mod schema_snapshot;
pub mod sql;
pub mod sql_safety;
pub mod storage;
+119
View File
@@ -0,0 +1,119 @@
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use crate::connection::AppState;
use crate::models::connection::{ConnectionConfig, DatabaseType};
use crate::{schema, types};
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct TableSnapshot {
pub name: String,
pub table_type: String,
pub comment: Option<String>,
pub columns: Vec<types::ColumnInfo>,
pub indexes: Vec<types::IndexInfo>,
pub foreign_keys: Vec<types::ForeignKeyInfo>,
pub triggers: Vec<types::TriggerInfo>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
#[serde(rename_all = "camelCase")]
pub struct SchemaSnapshot {
pub connection_id: String,
pub connection_name: String,
pub database: Option<String>,
pub database_type: DatabaseType,
pub driver_profile: Option<String>,
pub captured_at: DateTime<Utc>,
pub databases: Vec<types::DatabaseInfo>,
pub schemas: Vec<String>,
pub tables: Vec<TableSnapshot>,
pub views: Vec<TableSnapshot>,
}
pub async fn snapshot(
state: &AppState,
connection_id: &str,
database: Option<&str>,
schema_name: Option<&str>,
) -> Result<SchemaSnapshot, String> {
let config = {
let configs = state.configs.lock().await;
configs.get(connection_id).cloned().ok_or("Connection config not found")?
};
let db = snapshot_database(&config, database);
let databases = schema::list_databases_core(state, connection_id).await.unwrap_or_default();
let schemas = if db.is_empty() {
Vec::new()
} else {
schema::list_schemas_core(state, connection_id, &db).await.unwrap_or_default()
};
let effective_schema = schema_name.or_else(|| schemas.first().map(String::as_str)).unwrap_or("");
let table_infos = if db.is_empty() {
Vec::new()
} else {
schema::list_tables_core(state, connection_id, &db, effective_schema).await.unwrap_or_default()
};
let mut tables = Vec::new();
let mut views = Vec::new();
for table in table_infos {
let table_snapshot = TableSnapshot {
columns: schema::get_columns_core(state, connection_id, &db, effective_schema, &table.name)
.await
.unwrap_or_default(),
indexes: schema::list_indexes_core(state, connection_id, &db, effective_schema, &table.name)
.await
.unwrap_or_default(),
foreign_keys: schema::list_foreign_keys_core(state, connection_id, &db, effective_schema, &table.name)
.await
.unwrap_or_default(),
triggers: schema::list_triggers_core(state, connection_id, &db, effective_schema, &table.name)
.await
.unwrap_or_default(),
name: table.name,
table_type: table.table_type,
comment: table.comment,
};
if is_view(&table_snapshot.table_type) {
views.push(table_snapshot);
} else {
tables.push(table_snapshot);
}
}
Ok(SchemaSnapshot {
connection_id: config.id,
connection_name: config.name,
database: (!db.is_empty()).then_some(db),
database_type: config.db_type,
driver_profile: config.driver_profile,
captured_at: Utc::now(),
databases,
schemas,
tables,
views,
})
}
fn snapshot_database(config: &ConnectionConfig, requested_database: Option<&str>) -> String {
requested_database
.or_else(|| config.effective_database())
.or_else(|| embedded_default_database(&config.db_type))
.unwrap_or_default()
.to_string()
}
fn embedded_default_database(db_type: &DatabaseType) -> Option<&'static str> {
match db_type {
DatabaseType::Sqlite | DatabaseType::DuckDb => Some("main"),
_ => None,
}
}
fn is_view(table_type: &str) -> bool {
table_type.eq_ignore_ascii_case("view")
}
+99
View File
@@ -0,0 +1,99 @@
use std::str::FromStr;
use dbx_core::connection::{AppState, PoolKind};
use dbx_core::models::connection::{default_ssh_connect_timeout_secs, ConnectionConfig, DatabaseType};
use dbx_core::schema_snapshot::snapshot;
use dbx_core::storage::Storage;
use sqlx::sqlite::{SqliteConnectOptions, SqlitePoolOptions};
fn sqlite_config(path: &std::path::Path) -> ConnectionConfig {
ConnectionConfig {
id: "sqlite-fixture".to_string(),
name: "SQLite Fixture".to_string(),
db_type: DatabaseType::Sqlite,
driver_profile: Some("builtin-sqlite".to_string()),
driver_label: None,
url_params: None,
host: path.display().to_string(),
port: 0,
username: String::new(),
password: String::new(),
database: None,
color: None,
ssh_enabled: false,
ssh_host: String::new(),
ssh_port: 22,
ssh_user: String::new(),
ssh_password: String::new(),
ssh_key_path: String::new(),
ssh_key_passphrase: String::new(),
ssh_expose_lan: false,
ssh_connect_timeout_secs: default_ssh_connect_timeout_secs(),
ssl: false,
sysdba: false,
connection_string: None,
jdbc_driver_class: None,
jdbc_driver_paths: Vec::new(),
}
}
async fn create_sqlite_fixture(path: &std::path::Path) {
let url = format!("sqlite:{}?mode=rwc", path.display());
let options = SqliteConnectOptions::from_str(&url).unwrap().create_if_missing(true);
let pool = SqlitePoolOptions::new().max_connections(1).connect_with(options).await.unwrap();
sqlx::query("PRAGMA foreign_keys = ON").execute(&pool).await.unwrap();
sqlx::query("CREATE TABLE teams (id INTEGER PRIMARY KEY, name TEXT NOT NULL)").execute(&pool).await.unwrap();
sqlx::query(
"CREATE TABLE users (id INTEGER PRIMARY KEY, team_id INTEGER NOT NULL, email TEXT NOT NULL UNIQUE, FOREIGN KEY(team_id) REFERENCES teams(id))",
)
.execute(&pool)
.await
.unwrap();
sqlx::query("CREATE INDEX idx_users_team_id ON users(team_id)").execute(&pool).await.unwrap();
sqlx::query("CREATE TRIGGER trg_users_ai AFTER INSERT ON users BEGIN SELECT 1; END").execute(&pool).await.unwrap();
sqlx::query("CREATE VIEW active_users AS SELECT id, email FROM users").execute(&pool).await.unwrap();
pool.close().await;
}
async fn open_state() -> AppState {
let storage_path = std::env::temp_dir().join(format!("dbx-schema-snapshot-storage-{}.db", uuid::Uuid::new_v4()));
AppState::new(Storage::open(&storage_path).await.unwrap())
}
#[tokio::test]
async fn snapshot_standardizes_sqlite_tables_views_and_metadata() {
let data_path = std::env::temp_dir().join(format!("dbx-schema-snapshot-data-{}.db", uuid::Uuid::new_v4()));
create_sqlite_fixture(&data_path).await;
let state = open_state().await;
let config = sqlite_config(&data_path);
let pool = dbx_core::db::sqlite::connect_path(&data_path.display().to_string()).await.unwrap();
state.configs.lock().await.insert(config.id.clone(), config.clone());
state.connections.lock().await.insert(config.id.clone(), PoolKind::Sqlite(pool));
let snapshot = snapshot(&state, &config.id, None, None).await.unwrap();
assert_eq!(snapshot.connection_id, "sqlite-fixture");
assert_eq!(snapshot.connection_name, "SQLite Fixture");
assert_eq!(snapshot.database.as_deref(), Some("main"));
assert_eq!(snapshot.database_type, DatabaseType::Sqlite);
assert_eq!(snapshot.driver_profile.as_deref(), Some("builtin-sqlite"));
let now = chrono::Utc::now();
assert!(snapshot.captured_at <= now);
assert!(snapshot.captured_at > now - chrono::Duration::seconds(5));
assert_eq!(snapshot.databases.iter().map(|db| db.name.as_str()).collect::<Vec<_>>(), vec!["main"]);
let table_names = snapshot.tables.iter().map(|table| table.name.as_str()).collect::<Vec<_>>();
assert_eq!(table_names, vec!["teams", "users"]);
let view_names = snapshot.views.iter().map(|view| view.name.as_str()).collect::<Vec<_>>();
assert_eq!(view_names, vec!["active_users"]);
let users = snapshot.tables.iter().find(|table| table.name == "users").unwrap();
assert_eq!(users.table_type, "BASE TABLE");
assert!(users.columns.iter().any(|column| column.name == "email" && !column.is_nullable));
assert!(users.indexes.iter().any(|index| index.name == "idx_users_team_id" && index.columns == vec!["team_id"]));
assert!(users.foreign_keys.iter().any(|fk| fk.column == "team_id" && fk.ref_table == "teams"));
assert!(users.triggers.iter().any(|trigger| trigger.name == "trg_users_ai" && trigger.event == "INSERT"));
}