feat(plugin): 新增数据统计 REST API — count 和 aggregate 端点

- dynamic_table: 新增 build_filtered_count_sql(带过滤/搜索的 COUNT)和 build_aggregate_sql(按字段分组计数)
- data_service: 新增 count 和 aggregate 方法,支持实时统计查询
- data_handler: 新增 count_plugin_data 和 aggregate_plugin_data REST handler
- data_dto: 新增 AggregateItem、AggregateQueryParams、CountQueryParams 类型
- module: 注册 /plugins/{plugin_id}/{entity}/count 和 /aggregate 路由
- 包含 8 个新增单元测试,全部通过
This commit is contained in:
iven
2026-04-16 16:22:33 +08:00
parent 169e6d1fe5
commit 9effa9f942
6 changed files with 460 additions and 1 deletions

9
Cargo.lock generated
View File

@@ -1251,6 +1251,15 @@ dependencies = [
"wasmtime-wasi", "wasmtime-wasi",
] ]
[[package]]
name = "erp-plugin-crm"
version = "0.1.0"
dependencies = [
"serde",
"serde_json",
"wit-bindgen 0.55.0",
]
[[package]] [[package]]
name = "erp-plugin-prototype" name = "erp-plugin-prototype"
version = "0.1.0" version = "0.1.0"

View File

@@ -36,3 +36,30 @@ pub struct PluginDataListParams {
/// "asc" or "desc" /// "asc" or "desc"
pub sort_order: Option<String>, pub sort_order: Option<String>,
} }
/// 聚合查询响应项
#[derive(Debug, Serialize, Deserialize, utoipa::ToSchema)]
pub struct AggregateItem {
/// 分组键(字段值)
pub key: String,
/// 计数
pub count: i64,
}
/// 聚合查询参数
#[derive(Debug, Serialize, Deserialize, utoipa::IntoParams)]
pub struct AggregateQueryParams {
/// 分组字段名
pub group_by: String,
/// JSON 格式过滤: {"field":"value"}
pub filter: Option<String>,
}
/// 统计查询参数
#[derive(Debug, Serialize, Deserialize, utoipa::IntoParams)]
pub struct CountQueryParams {
/// 搜索关键词
pub search: Option<String>,
/// JSON 格式过滤: {"field":"value"}
pub filter: Option<String>,
}

View File

@@ -261,6 +261,99 @@ impl PluginDataService {
Ok(()) Ok(())
} }
/// 统计记录数(支持过滤和搜索)
pub async fn count(
plugin_id: Uuid,
entity_name: &str,
tenant_id: Uuid,
db: &sea_orm::DatabaseConnection,
filter: Option<serde_json::Value>,
search: Option<String>,
) -> AppResult<u64> {
let table_name = resolve_table_name(plugin_id, entity_name, tenant_id, db).await?;
// 获取 searchable 字段列表,构建搜索条件
let entity_fields = resolve_entity_fields(plugin_id, entity_name, tenant_id, db).await?;
let search_tuple = {
let searchable: Vec<&str> = entity_fields
.iter()
.filter(|f| f.searchable == Some(true))
.map(|f| f.name.as_str())
.collect();
match (searchable.is_empty(), &search) {
(false, Some(kw)) => Some((searchable.join(","), kw.clone())),
_ => None,
}
};
let (sql, values) = DynamicTableManager::build_filtered_count_sql(
&table_name,
tenant_id,
filter,
search_tuple,
)
.map_err(|e| AppError::Validation(e))?;
#[derive(FromQueryResult)]
struct CountResult {
count: i64,
}
let result = CountResult::find_by_statement(Statement::from_sql_and_values(
sea_orm::DatabaseBackend::Postgres,
sql,
values,
))
.one(db)
.await?
.map(|r| r.count as u64)
.unwrap_or(0);
Ok(result)
}
/// 聚合查询 — 按字段分组计数
/// 返回 [(分组键, 计数), ...]
pub async fn aggregate(
plugin_id: Uuid,
entity_name: &str,
tenant_id: Uuid,
db: &sea_orm::DatabaseConnection,
group_by_field: &str,
filter: Option<serde_json::Value>,
) -> AppResult<Vec<(String, i64)>> {
let table_name = resolve_table_name(plugin_id, entity_name, tenant_id, db).await?;
let (sql, values) = DynamicTableManager::build_aggregate_sql(
&table_name,
tenant_id,
group_by_field,
filter,
)
.map_err(|e| AppError::Validation(e))?;
#[derive(FromQueryResult)]
struct AggRow {
key: Option<String>,
count: i64,
}
let rows = AggRow::find_by_statement(Statement::from_sql_and_values(
sea_orm::DatabaseBackend::Postgres,
sql,
values,
))
.all(db)
.await?;
let result = rows
.into_iter()
.map(|r| (r.key.unwrap_or_default(), r.count))
.collect();
Ok(result)
}
} }
/// 从 plugin_entities 表解析 table_name带租户隔离 /// 从 plugin_entities 表解析 table_name带租户隔离

View File

@@ -287,6 +287,117 @@ impl DynamicTableManager {
) )
} }
/// 构建带过滤条件的 COUNT SQL
/// 复用 build_filtered_query_sql 的条件构建逻辑,但只做 COUNT
pub fn build_filtered_count_sql(
table_name: &str,
tenant_id: Uuid,
filter: Option<serde_json::Value>,
search: Option<(String, String)>, // (searchable_fields_csv, keyword)
) -> Result<(String, Vec<Value>), String> {
let mut conditions = vec![
format!("\"tenant_id\" = ${}", 1),
"\"deleted_at\" IS NULL".to_string(),
];
let mut param_idx = 2;
let mut values: Vec<Value> = vec![tenant_id.into()];
// 处理 filter与 build_filtered_query_sql 保持一致)
if let Some(f) = filter {
if let Some(obj) = f.as_object() {
for (key, val) in obj {
let clean_key = sanitize_identifier(key);
if clean_key.is_empty() {
return Err(format!("无效的过滤字段名: {}", key));
}
conditions.push(format!("\"data\"->>'{}' = ${}", clean_key, param_idx));
values.push(Value::String(Some(Box::new(
val.as_str().unwrap_or("").to_string(),
))));
param_idx += 1;
}
}
}
// 处理 search与 build_filtered_query_sql 保持一致)
if let Some((fields_csv, keyword)) = search {
let escaped = keyword.replace('%', "\\%").replace('_', "\\_");
let fields: Vec<&str> = fields_csv.split(',').collect();
let search_param_idx = param_idx;
let search_conditions: Vec<String> = fields
.iter()
.map(|f| {
let clean = sanitize_identifier(f.trim());
format!("\"data\"->>'{}' ILIKE ${}", clean, search_param_idx)
})
.collect();
conditions.push(format!("({})", search_conditions.join(" OR ")));
values.push(Value::String(Some(Box::new(format!("%{}%", escaped)))));
}
let sql = format!(
"SELECT COUNT(*) as count FROM \"{}\" WHERE {}",
table_name,
conditions.join(" AND "),
);
Ok((sql, values))
}
/// 构建聚合查询 SQL — 按 JSONB 字段分组计数
/// SELECT data->>'group_field' as key, COUNT(*) as count
/// FROM table WHERE tenant_id = $1 AND deleted_at IS NULL [AND filter...]
/// GROUP BY data->>'group_field' ORDER BY count DESC
pub fn build_aggregate_sql(
table_name: &str,
tenant_id: Uuid,
group_by_field: &str,
filter: Option<serde_json::Value>,
) -> Result<(String, Vec<Value>), String> {
let clean_group = sanitize_identifier(group_by_field);
if clean_group.is_empty() {
return Err(format!("无效的分组字段名: {}", group_by_field));
}
let mut conditions = vec![
format!("\"tenant_id\" = ${}", 1),
"\"deleted_at\" IS NULL".to_string(),
];
let mut param_idx = 2;
let mut values: Vec<Value> = vec![tenant_id.into()];
// 处理 filter
if let Some(f) = filter {
if let Some(obj) = f.as_object() {
for (key, val) in obj {
let clean_key = sanitize_identifier(key);
if clean_key.is_empty() {
return Err(format!("无效的过滤字段名: {}", key));
}
conditions.push(format!("\"data\"->>'{}' = ${}", clean_key, param_idx));
values.push(Value::String(Some(Box::new(
val.as_str().unwrap_or("").to_string(),
))));
param_idx += 1;
}
}
}
let sql = format!(
"SELECT \"data\"->>'{}' as key, COUNT(*) as count \
FROM \"{}\" \
WHERE {} \
GROUP BY \"data\"->>'{}' \
ORDER BY count DESC",
clean_group,
table_name,
conditions.join(" AND "),
clean_group,
);
Ok((sql, values))
}
/// 构建带过滤条件的查询 SQL /// 构建带过滤条件的查询 SQL
pub fn build_filtered_query_sql( pub fn build_filtered_query_sql(
table_name: &str, table_name: &str,
@@ -492,4 +603,116 @@ mod tests {
sql sql
); );
} }
// ===== build_filtered_count_sql 测试 =====
#[test]
fn test_build_filtered_count_sql_basic() {
let (sql, values) = DynamicTableManager::build_filtered_count_sql(
"plugin_test",
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
None,
None,
)
.unwrap();
assert!(sql.contains("COUNT(*)"), "Expected COUNT(*), got: {}", sql);
assert!(sql.contains("tenant_id"), "Expected tenant_id filter");
assert!(sql.contains("deleted_at"), "Expected soft delete filter");
assert_eq!(values.len(), 1); // 仅 tenant_id
}
#[test]
fn test_build_filtered_count_sql_with_filter() {
let (sql, values) = DynamicTableManager::build_filtered_count_sql(
"plugin_test_customer",
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
Some(serde_json::json!({"status": "active"})),
None,
)
.unwrap();
assert!(sql.contains("\"data\"->>'status' ="), "Expected filter, got: {}", sql);
assert_eq!(values.len(), 2); // tenant_id + filter_value
}
#[test]
fn test_build_filtered_count_sql_with_search() {
let (sql, values) = DynamicTableManager::build_filtered_count_sql(
"plugin_test",
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
None,
Some(("name,code".to_string(), "搜索词".to_string())),
)
.unwrap();
assert!(sql.contains("ILIKE"), "Expected ILIKE, got: {}", sql);
assert_eq!(values.len(), 2); // tenant_id + search_param
}
#[test]
fn test_build_filtered_count_sql_no_limit_offset() {
// COUNT SQL 不应包含 LIMIT/OFFSET
let (sql, _) = DynamicTableManager::build_filtered_count_sql(
"plugin_test",
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
None,
None,
)
.unwrap();
assert!(!sql.contains("LIMIT"), "COUNT SQL 不应包含 LIMIT");
assert!(!sql.contains("OFFSET"), "COUNT SQL 不应包含 OFFSET");
}
// ===== build_aggregate_sql 测试 =====
#[test]
fn test_build_aggregate_sql_basic() {
let (sql, values) = DynamicTableManager::build_aggregate_sql(
"plugin_test_customer",
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
"status",
None,
)
.unwrap();
assert!(sql.contains("GROUP BY"), "Expected GROUP BY, got: {}", sql);
assert!(sql.contains("\"data\"->>'status'"), "Expected group field, got: {}", sql);
assert!(sql.contains("ORDER BY count DESC"), "Expected ORDER BY count DESC, got: {}", sql);
assert_eq!(values.len(), 1); // 仅 tenant_id
}
#[test]
fn test_build_aggregate_sql_with_filter() {
let (sql, values) = DynamicTableManager::build_aggregate_sql(
"plugin_test_customer",
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
"region",
Some(serde_json::json!({"status": "active"})),
)
.unwrap();
assert!(sql.contains("\"data\"->>'region'"), "Expected group field, got: {}", sql);
assert!(sql.contains("\"data\"->>'status' ="), "Expected filter, got: {}", sql);
assert_eq!(values.len(), 2); // tenant_id + filter_value
}
#[test]
fn test_build_aggregate_sql_sanitizes_group_field() {
let result = DynamicTableManager::build_aggregate_sql(
"plugin_test",
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
"evil'; DROP TABLE--",
None,
);
let (sql, _) = result.unwrap();
assert!(!sql.contains("DROP TABLE"), "SQL 不应包含注入: {}", sql);
assert!(sql.contains("evil___DROP_TABLE__"), "字段名应被清理: {}", sql);
}
#[test]
fn test_build_aggregate_sql_empty_field_rejected() {
let result = DynamicTableManager::build_aggregate_sql(
"plugin_test",
Uuid::parse_str("00000000-0000-0000-0000-000000000001").unwrap(),
"",
None,
);
assert!(result.is_err(), "空字段名应被拒绝");
}
} }

View File

@@ -7,7 +7,10 @@ use erp_core::error::AppError;
use erp_core::rbac::require_permission; use erp_core::rbac::require_permission;
use erp_core::types::{ApiResponse, PaginatedResponse, TenantContext}; use erp_core::types::{ApiResponse, PaginatedResponse, TenantContext};
use crate::data_dto::{CreatePluginDataReq, PluginDataListParams, PluginDataResp, UpdatePluginDataReq}; use crate::data_dto::{
AggregateItem, AggregateQueryParams, CountQueryParams, CreatePluginDataReq,
PluginDataListParams, PluginDataResp, UpdatePluginDataReq,
};
use crate::data_service::PluginDataService; use crate::data_service::PluginDataService;
use crate::state::PluginState; use crate::state::PluginState;
@@ -228,3 +231,98 @@ where
Ok(Json(ApiResponse::ok(()))) Ok(Json(ApiResponse::ok(())))
} }
#[utoipa::path(
get,
path = "/api/v1/plugins/{plugin_id}/{entity}/count",
params(CountQueryParams),
responses(
(status = 200, description = "成功", body = ApiResponse<u64>),
),
security(("bearer_auth" = [])),
tag = "插件数据"
)]
/// GET /api/v1/plugins/{plugin_id}/{entity}/count — 统计计数
pub async fn count_plugin_data<S>(
State(state): State<PluginState>,
Extension(ctx): Extension<TenantContext>,
Path((plugin_id, entity)): Path<(Uuid, String)>,
Query(params): Query<CountQueryParams>,
) -> Result<Json<ApiResponse<u64>>, AppError>
where
PluginState: FromRef<S>,
S: Clone + Send + Sync + 'static,
{
let fine_perm = compute_permission_code(&plugin_id.to_string(), &entity, "list");
if require_permission(&ctx, &fine_perm).is_err() {
require_permission(&ctx, "plugin.list")?;
}
// 解析 filter JSON
let filter: Option<serde_json::Value> = params
.filter
.as_ref()
.and_then(|f| serde_json::from_str(f).ok());
let total = PluginDataService::count(
plugin_id,
&entity,
ctx.tenant_id,
&state.db,
filter,
params.search,
)
.await?;
Ok(Json(ApiResponse::ok(total)))
}
#[utoipa::path(
get,
path = "/api/v1/plugins/{plugin_id}/{entity}/aggregate",
params(AggregateQueryParams),
responses(
(status = 200, description = "成功", body = ApiResponse<Vec<AggregateItem>>),
),
security(("bearer_auth" = [])),
tag = "插件数据"
)]
/// GET /api/v1/plugins/{plugin_id}/{entity}/aggregate — 聚合查询
pub async fn aggregate_plugin_data<S>(
State(state): State<PluginState>,
Extension(ctx): Extension<TenantContext>,
Path((plugin_id, entity)): Path<(Uuid, String)>,
Query(params): Query<AggregateQueryParams>,
) -> Result<Json<ApiResponse<Vec<AggregateItem>>>, AppError>
where
PluginState: FromRef<S>,
S: Clone + Send + Sync + 'static,
{
let fine_perm = compute_permission_code(&plugin_id.to_string(), &entity, "list");
if require_permission(&ctx, &fine_perm).is_err() {
require_permission(&ctx, "plugin.list")?;
}
// 解析 filter JSON
let filter: Option<serde_json::Value> = params
.filter
.as_ref()
.and_then(|f| serde_json::from_str(f).ok());
let rows = PluginDataService::aggregate(
plugin_id,
&entity,
ctx.tenant_id,
&state.db,
&params.group_by,
filter,
)
.await?;
let items = rows
.into_iter()
.map(|(key, count)| AggregateItem { key, count })
.collect();
Ok(Json(ApiResponse::ok(items)))
}

View File

@@ -76,6 +76,15 @@ impl PluginModule {
get(crate::handler::data_handler::get_plugin_data::<S>) get(crate::handler::data_handler::get_plugin_data::<S>)
.put(crate::handler::data_handler::update_plugin_data::<S>) .put(crate::handler::data_handler::update_plugin_data::<S>)
.delete(crate::handler::data_handler::delete_plugin_data::<S>), .delete(crate::handler::data_handler::delete_plugin_data::<S>),
)
// 数据统计路由
.route(
"/plugins/{plugin_id}/{entity}/count",
get(crate::handler::data_handler::count_plugin_data::<S>),
)
.route(
"/plugins/{plugin_id}/{entity}/aggregate",
get(crate::handler::data_handler::aggregate_plugin_data::<S>),
); );
admin_routes.merge(data_routes) admin_routes.merge(data_routes)