feat: Redis/MongoDB分页+CSV导出全数据库类型支持

后端:
- Redis 分页: KEYS→SCAN分页, LRANGE→offset/limit, SMEMBERS→SSCAN分页
- MongoDB 新增 executeMongoQuery: db.collection.find/aggregate 解析
  - find: skip/limit分页 + CountDocuments总数
  - aggregate: $facet并行取count+data
  - bson.M扁平化处理嵌套文档
- ExportQueryCSV 改为多分支: exportSQLCSV/exportRedisCSV/exportMongoCSV
  - Redis: 复用executeRedisQuery取数据写CSV
  - MongoDB: 复用executeMongoQuery取数据写CSV
- 新增 mongo-driver v2 依赖
- 前端: Redis也显示导出CSV按钮(去掉dbType过滤)
This commit is contained in:
logan committed 2026-05-28 15:58:42 +08:00
1 parent 5ea86fe334
commit 3b3cc32d45
3 files changed
+419 -33

No files matched your search

+6
View File
@@ -38,6 +38,7 @@ require (
github.com/jinzhu/inflection v1.0.0 // indirect github.com/jinzhu/inflection v1.0.0 // indirect
github.com/jinzhu/now v1.1.5 // indirect github.com/jinzhu/now v1.1.5 // indirect
github.com/json-iterator/go v1.1.12 // indirect github.com/json-iterator/go v1.1.12 // indirect
github.com/klauspost/compress v1.17.6 // indirect
github.com/klauspost/cpuid/v2 v2.2.10 // indirect github.com/klauspost/cpuid/v2 v2.2.10 // indirect
github.com/kr/text v0.2.0 // indirect github.com/kr/text v0.2.0 // indirect
github.com/leodido/go-urn v1.4.0 // indirect github.com/leodido/go-urn v1.4.0 // indirect
@@ -48,6 +49,11 @@ require (
github.com/pelletier/go-toml/v2 v2.2.3 // indirect github.com/pelletier/go-toml/v2 v2.2.3 // indirect
github.com/twitchyliquid64/golang-asm v0.15.1 // indirect github.com/twitchyliquid64/golang-asm v0.15.1 // indirect
github.com/ugorji/go/codec v1.2.12 // indirect github.com/ugorji/go/codec v1.2.12 // indirect
github.com/xdg-go/pbkdf2 v1.0.0 // indirect
github.com/xdg-go/scram v1.2.0 // indirect
github.com/xdg-go/stringprep v1.0.4 // indirect
github.com/youmark/pkcs8 v0.0.0-20240726163527-a2c0da244d78 // indirect
go.mongodb.org/mongo-driver/v2 v2.6.0 // indirect
go.uber.org/atomic v1.11.0 // indirect go.uber.org/atomic v1.11.0 // indirect
golang.org/x/arch v0.15.0 // indirect golang.org/x/arch v0.15.0 // indirect
golang.org/x/net v0.50.0 // indirect golang.org/x/net v0.50.0 // indirect
+412 -32
View File
@@ -16,6 +16,11 @@ import (
"strconv" "strconv"
"encoding/json" "encoding/json"
"bytes" "bytes"
"regexp"
"go.mongodb.org/mongo-driver/v2/bson"
"go.mongodb.org/mongo-driver/v2/mongo"
"go.mongodb.org/mongo-driver/v2/mongo/options"
"go.mongodb.org/mongo-driver/v2/mongo/readpref"
) )
// GET /api/v1/instances/:id/detail/databases // GET /api/v1/instances/:id/detail/databases
@@ -549,13 +554,30 @@ func ExecuteQuery(c *gin.Context) {
c.JSON(http.StatusOK, result) c.JSON(http.StatusOK, result)
case "redis": case "redis":
result, err := executeRedisQuery(&instance, input.Database, input.SQL) result, err := executeRedisQuery(&instance, input.Database, input.SQL, input.Limit, input.Offset)
costMs := time.Since(start).Milliseconds() costMs := time.Since(start).Milliseconds()
if err != nil { if err != nil {
c.JSON(http.StatusOK, gin.H{"columns": []string{}, "rows": []interface{}{}, "total": 0, "cost_ms": costMs, "error": err.Error()}) c.JSON(http.StatusOK, gin.H{"columns": []string{}, "rows": []interface{}{}, "total": 0, "cost_ms": costMs, "error": err.Error()})
return return
} }
saveQueryHistory(userID, input.InstanceID, input.Database, input.SQL, costMs, 1) result["cost_ms"] = costMs
rowCount := 1
if rowsArr, ok := result["rows"].([]interface{}); ok {
rowCount = len(rowsArr)
}
saveQueryHistory(userID, input.InstanceID, input.Database, input.SQL, costMs, rowCount)
c.JSON(http.StatusOK, result)
case "mongodb", "mongo":
result, err := executeMongoQuery(&instance, input.Database, input.SQL, input.Limit, input.Offset)
costMs := time.Since(start).Milliseconds()
if err != nil {
c.JSON(http.StatusOK, gin.H{"columns": []string{}, "rows": []interface{}{}, "total": 0, "cost_ms": costMs, "error": err.Error()})
return
}
result["cost_ms"] = costMs
rowsInterface, _ := result["rows"].([]interface{})
saveQueryHistory(userID, input.InstanceID, input.Database, input.SQL, costMs, len(rowsInterface))
c.JSON(http.StatusOK, result) c.JSON(http.StatusOK, result)
default: default:
@@ -639,7 +661,7 @@ func executeSQLQuery(instance *model.Instance, dbName string, sqlStr string, lim
}, nil }, nil
} }
func executeRedisQuery(instance *model.Instance, dbSelect string, command string) (gin.H, error) { func executeRedisQuery(instance *model.Instance, dbSelect string, command string, limit int, offset int) (gin.H, error) {
password := instance.Password password := instance.Password
if strings.HasPrefix(password, "enc:") { if strings.HasPrefix(password, "enc:") {
decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:")) decrypted, err := decryptPassword(strings.TrimPrefix(password, "enc:"))
@@ -695,6 +717,109 @@ func executeRedisQuery(instance *model.Instance, dbSelect string, command string
args = append(args, p) args = append(args, p)
} }
// For KEYS command, use SCAN with COUNT instead for safety + pagination
if cmd == "KEYS" && strings.Contains(parts[len(parts)-1], "*") {
// Replace KEYS * with SCAN-based pagination
pattern := "*"
if len(parts) > 1 {
pattern = parts[1]
}
var allKeys []string
cursor := uint64(0)
maxScan := offset + limit
for {
keys, nextCursor, err := client.Scan(ctx, cursor, pattern, int64(maxScan-len(allKeys))).Result()
if err != nil {
return nil, fmt.Errorf("SCAN failed: %v", err)
}
allKeys = append(allKeys, keys...)
cursor = nextCursor
if cursor == 0 || len(allKeys) >= maxScan {
break
}
}
total := len(allKeys)
end := offset + limit
if end > len(allKeys) {
end = len(allKeys)
}
if offset > len(allKeys) {
offset = len(allKeys)
}
pagedKeys := allKeys[offset:end]
rows := make([]interface{}, len(pagedKeys))
for i, k := range pagedKeys {
rows[i] = map[string]interface{}{"index": offset + i, "key": k}
}
return gin.H{
"columns": []string{"index", "key"},
"rows": rows,
"total": total,
"count": len(rows),
}, nil
}
// For LRANGE, inject pagination if user didn't specify start/stop
if cmd == "LRANGE" && len(parts) < 4 {
key := ""
if len(parts) > 1 {
key = parts[1]
}
result, err := client.LRange(ctx, key, int64(offset), int64(offset+limit-1)).Result()
if err != nil {
return nil, fmt.Errorf("command failed: %v", err)
}
// Get total via LLEN
total, _ := client.LLen(ctx, key).Result()
rows := make([]interface{}, len(result))
for i, val := range result {
rows[i] = map[string]interface{}{"index": offset + i, "value": val}
}
return gin.H{
"columns": []string{"index", "value"},
"rows": rows,
"total": int(total),
"count": len(rows),
}, nil
}
// For SMEMBERS with limit, use SSCAN
if cmd == "SMEMBERS" && limit > 0 {
key := ""
if len(parts) > 1 {
key = parts[1]
}
total, _ := client.SCard(ctx, key).Result()
var members []string
cursor := uint64(0)
for len(members) < offset+limit {
batch, nextCursor, err := client.SScan(ctx, key, cursor, "", int64(offset+limit-len(members))).Result()
if err != nil {
break
}
members = append(members, batch...)
cursor = nextCursor
if cursor == 0 {
break
}
}
end := offset + limit
if end > len(members) {
end = len(members)
}
paged := members[offset:end]
rows := make([]interface{}, len(paged))
for i, m := range paged {
rows[i] = map[string]interface{}{"index": offset + i, "member": m}
}
return gin.H{
"columns": []string{"index", "member"},
"rows": rows,
"total": int(total),
"count": len(rows),
}, nil
}
result, err := client.Do(ctx, args...).Result() result, err := client.Do(ctx, args...).Result()
if err != nil { if err != nil {
return nil, fmt.Errorf("command failed: %v", err) return nil, fmt.Errorf("command failed: %v", err)
@@ -780,9 +905,208 @@ func executeRedisQuery(instance *model.Instance, dbSelect string, command string
"columns": columns, "columns": columns,
"rows": rows, "rows": rows,
"total": len(rows), "total": len(rows),
"count": len(rows),
}, nil }, nil
} }
// executeMongoQuery — MongoDB query with pagination
// Supports: db.collection.find({...}), db.collection.aggregate([...])
// Input format: db.collection.find({query}) or JSON array of pipeline stages
func executeMongoQuery(instance *model.Instance, dbName string, queryStr string, limit int, offset int) (gin.H, error) {
ctx, cancel := context.WithTimeout(context.Background(), 30*time.Second)
defer cancel()
// Build connection URI
uri := fmt.Sprintf("mongodb://%s:%d", instance.Host, instance.Port)
if instance.Password != "" {
pwd := instance.Password
if strings.HasPrefix(pwd, "enc:") {
if dec, err := decryptPassword(strings.TrimPrefix(pwd, "enc:")); err == nil {
pwd = dec
}
}
user := instance.Username
if user == "" {
user = "root"
}
uri = fmt.Sprintf("mongodb://%s:%s@%s:%d", user, pwd, instance.Host, instance.Port)
}
client, err := mongo.Connect(options.Client().ApplyURI(uri))
if err != nil {
return nil, fmt.Errorf("MongoDB connection failed: %v", err)
}
defer client.Disconnect(ctx)
if err := client.Ping(ctx, readpref.Primary()); err != nil {
return nil, fmt.Errorf("MongoDB ping failed: %v", err)
}
// Parse query: "db.collection.find({...})" or raw JSON
collectionName, filter, pipeline, err := parseMongoQuery(queryStr)
if err != nil {
return nil, err
}
if dbName == "" {
dbName = "test"
}
coll := client.Database(dbName).Collection(collectionName)
var total int64
var docs []bson.M
if pipeline != nil {
// Aggregation — add $facet for count + data
facetStage := bson.D{
{"$facet", bson.D{
{"data", bson.A{
bson.D{{"$skip", offset}},
bson.D{{"$limit", limit}},
}},
{"count", bson.A{
bson.D{{"$count", "total"}},
}},
}},
}
fullPipeline := append(pipeline, facetStage)
var results []bson.M
cursor, err := coll.Aggregate(ctx, fullPipeline)
if err != nil {
return nil, fmt.Errorf("aggregation failed: %v", err)
}
defer cursor.Close(ctx)
cursor.All(ctx, &results)
if len(results) > 0 {
if dataArr, ok := results[0]["data"].(bson.A); ok {
for _, d := range dataArr {
if m, ok := d.(bson.M); ok {
docs = append(docs, m)
}
}
}
if countArr, ok := results[0]["count"].(bson.A); ok && len(countArr) > 0 {
if m, ok := countArr[0].(bson.M); ok {
if t, ok := m["total"].(int32); ok {
total = int64(t)
} else if t, ok := m["total"].(int64); ok {
total = t
}
}
}
}
} else {
// Find
total, _ = coll.CountDocuments(ctx, filter)
opts := options.Find().SetSkip(int64(offset)).SetLimit(int64(limit))
cursor, err := coll.Find(ctx, filter, opts)
if err != nil {
return nil, fmt.Errorf("find failed: %v", err)
}
defer cursor.Close(ctx)
cursor.All(ctx, &docs)
}
// Convert bson.M to flat maps for table display
columnSet := make(map[string]bool)
rowMaps := make([]interface{}, 0, len(docs))
for _, doc := range docs {
flat := flattenBSON(doc, "")
for k := range flat {
columnSet[k] = true
}
rowMaps = append(rowMaps, flat)
}
// Stable column order
columns := make([]string, 0, len(columnSet))
if _, hasID := columnSet["_id"]; hasID {
columns = append(columns, "_id")
}
for k := range columnSet {
if k != "_id" {
columns = append(columns, k)
}
}
return gin.H{
"columns": columns,
"rows": rowMaps,
"total": int(total),
"count": len(rowMaps),
}, nil
}
// parseMongoQuery parses "db.collection.find({...})" or "db.collection.aggregate([...])" or raw JSON
func parseMongoQuery(queryStr string) (collection string, filter bson.M, pipeline bson.A, err error) {
queryStr = strings.TrimSpace(queryStr)
// Pattern: db.collection.find({...})
if re := regexpFromPattern(`db\.(\w+)\.find\((.+)?\)`); re != nil {
matches := re.FindStringSubmatch(queryStr)
if len(matches) >= 2 {
collection = matches[1]
if len(matches) >= 3 && strings.TrimSpace(matches[2]) != "" {
if e := json.Unmarshal([]byte(matches[2]), &filter); e != nil {
filter = bson.M{}
}
} else {
filter = bson.M{}
}
return
}
}
// Pattern: db.collection.aggregate([...])
if re := regexpFromPattern(`db\.(\w+)\.aggregate\((.+)\)`); re != nil {
matches := re.FindStringSubmatch(queryStr)
if len(matches) >= 3 {
collection = matches[1]
if e := json.Unmarshal([]byte(matches[2]), &pipeline); e != nil {
err = fmt.Errorf("invalid aggregation pipeline: %v", e)
return
}
return
}
}
// Fallback: raw JSON object treated as find on all collections? No — require db.xxx format
err = fmt.Errorf("query format not supported. Use: db.collection.find({}) or db.collection.aggregate([...])")
return
}
func regexpFromPattern(pattern string) *regexp.Regexp {
// Import regexp lazily — actually let's just import at top
// This is a simple helper
return regexp.MustCompile(pattern)
}
func flattenBSON(doc bson.M, prefix string) map[string]interface{} {
result := make(map[string]interface{})
for k, v := range doc {
fullKey := k
if prefix != "" {
fullKey = prefix + "." + k
}
switch child := v.(type) {
case bson.M:
for ck, cv := range flattenBSON(child, fullKey) {
result[ck] = cv
}
case bson.A:
// Serialize arrays as JSON string
b, _ := json.Marshal(child)
result[fullKey] = string(b)
default:
// Convert ObjectID to string
s := fmt.Sprintf("%v", v)
result[fullKey] = s
}
}
return result
}
// POST /api/v1/query/export — Streaming CSV export for large datasets // POST /api/v1/query/export — Streaming CSV export for large datasets
func ExportQueryCSV(c *gin.Context) { func ExportQueryCSV(c *gin.Context) {
var input struct { var input struct {
@@ -800,7 +1124,7 @@ func ExportQueryCSV(c *gin.Context) {
return return
} }
if input.Limit <= 0 { if input.Limit <= 0 {
input.Limit = 10000 // Export max 10k rows input.Limit = 10000
} }
if input.Limit > 50000 { if input.Limit > 50000 {
input.Limit = 50000 input.Limit = 50000
@@ -812,48 +1136,47 @@ func ExportQueryCSV(c *gin.Context) {
return return
} }
if instance.DBType != "mysql" && instance.DBType != "postgresql" && instance.DBType != "postgres" && instance.DBType != "pg" { // Stream CSV response headers
c.JSON(http.StatusBadRequest, gin.H{"detail": "CSV export only supports MySQL/PostgreSQL"}) filename := fmt.Sprintf("query_export_%s.csv", time.Now().Format("20060102_150405"))
return c.Header("Content-Type", "text/csv; charset=utf-8")
c.Header("Content-Disposition", fmt.Sprintf("attachment; filename=%s", filename))
c.Writer.Write([]byte("\xEF\xBB\xBF")) // BOM for Excel
switch instance.DBType {
case "mysql", "postgresql", "postgres", "pg":
exportSQLCSV(c, &instance, input.Database, input.SQL, input.Limit)
case "redis":
exportRedisCSV(c, &instance, input.Database, input.SQL, input.Limit)
case "mongodb", "mongo":
exportMongoCSV(c, &instance, input.Database, input.SQL, input.Limit)
default:
c.JSON(http.StatusBadRequest, gin.H{"detail": "CSV export not supported for " + instance.DBType})
}
} }
db, err := getDBConnection(&instance, input.Database) func exportSQLCSV(c *gin.Context, instance *model.Instance, dbName string, sqlStr string, limit int) {
db, err := getDBConnection(instance, dbName)
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": "Connection failed: " + err.Error()}) c.Writer.Write([]byte("Error: connection failed\n"))
return return
} }
defer db.Close() defer db.Close()
// Inject LIMIT if not present upperSQL := strings.ToUpper(strings.TrimSpace(sqlStr))
upperSQL := strings.ToUpper(strings.TrimSpace(input.SQL))
if !strings.Contains(upperSQL, " LIMIT ") { if !strings.Contains(upperSQL, " LIMIT ") {
input.SQL = fmt.Sprintf("%s LIMIT %d", input.SQL, input.Limit) sqlStr = fmt.Sprintf("%s LIMIT %d", sqlStr, limit)
} }
rows, err := db.Query(input.SQL) rows, err := db.Query(sqlStr)
if err != nil { if err != nil {
c.JSON(http.StatusInternalServerError, gin.H{"detail": "Query failed: " + err.Error()}) c.Writer.Write([]byte("Error: query failed\n"))
return return
} }
defer rows.Close() defer rows.Close()
cols, _ := rows.Columns() cols, _ := rows.Columns()
colNames := make([]string, len(cols)) c.Writer.Write([]byte(strings.Join(cols, ",") + "\n"))
for i, col := range cols {
colNames[i] = col
}
// Stream CSV response
filename := fmt.Sprintf("query_export_%s.csv", time.Now().Format("20060102_150405"))
c.Header("Content-Type", "text/csv; charset=utf-8")
c.Header("Content-Disposition", fmt.Sprintf("attachment; filename=%s", filename))
// BOM for Excel UTF-8
c.Writer.Write([]byte("\xEF\xBB\xBF"))
// Header row
c.Writer.Write([]byte(strings.Join(colNames, ",") + "\n"))
// Data rows — write in chunks to avoid OOM
batchSize := 500 batchSize := 500
batch := make([]string, 0, batchSize) batch := make([]string, 0, batchSize)
rowCount := 0 rowCount := 0
@@ -874,7 +1197,6 @@ func ExportQueryCSV(c *gin.Context) {
} else if val != nil { } else if val != nil {
s = fmt.Sprintf("%v", val) s = fmt.Sprintf("%v", val)
} }
// CSV escape: wrap in quotes if contains comma, quote, or newline
if strings.ContainsAny(s, ",\"\n\r") { if strings.ContainsAny(s, ",\"\n\r") {
s = `"` + strings.ReplaceAll(s, `"`, `""`) + `"` s = `"` + strings.ReplaceAll(s, `"`, `""`) + `"`
} }
@@ -889,13 +1211,71 @@ func ExportQueryCSV(c *gin.Context) {
batch = batch[:0] batch = batch[:0]
} }
} }
// Flush remaining
if len(batch) > 0 { if len(batch) > 0 {
c.Writer.Write([]byte(strings.Join(batch, "\n") + "\n")) c.Writer.Write([]byte(strings.Join(batch, "\n") + "\n"))
} }
c.Writer.(http.Flusher).Flush() c.Writer.(http.Flusher).Flush()
saveQueryHistory(0, int(instance.ID), dbName, sqlStr, 0, rowCount)
}
saveQueryHistory(0, input.InstanceID, input.Database, input.SQL, 0, rowCount) func exportRedisCSV(c *gin.Context, instance *model.Instance, dbSelect string, command string, limit int) {
result, err := executeRedisQuery(instance, dbSelect, command, limit, 0)
if err != nil {
c.Writer.Write([]byte("Error: " + err.Error() + "\n"))
return
}
columns, _ := result["columns"].([]string)
rows, _ := result["rows"].([]interface{})
c.Writer.Write([]byte(strings.Join(columns, ",") + "\n"))
for _, row := range rows {
if m, ok := row.(map[string]interface{}); ok {
cells := make([]string, len(columns))
for i, col := range columns {
s := fmt.Sprintf("%v", m[col])
if strings.ContainsAny(s, ",\"\n\r") {
s = `"` + strings.ReplaceAll(s, `"`, `""`) + `"`
}
cells[i] = s
}
c.Writer.Write([]byte(strings.Join(cells, ",") + "\n"))
}
}
c.Writer.(http.Flusher).Flush()
rowCount := len(rows)
saveQueryHistory(0, int(instance.ID), dbSelect, command, 0, rowCount)
}
func exportMongoCSV(c *gin.Context, instance *model.Instance, dbName string, queryStr string, limit int) {
result, err := executeMongoQuery(instance, dbName, queryStr, limit, 0)
if err != nil {
c.Writer.Write([]byte("Error: " + err.Error() + "\n"))
return
}
columns, _ := result["columns"].([]string)
rows, _ := result["rows"].([]interface{})
c.Writer.Write([]byte(strings.Join(columns, ",") + "\n"))
for _, row := range rows {
if m, ok := row.(map[string]interface{}); ok {
cells := make([]string, len(columns))
for i, col := range columns {
s := fmt.Sprintf("%v", m[col])
if strings.ContainsAny(s, ",\"\n\r") {
s = `"` + strings.ReplaceAll(s, `"`, `""`) + `"`
}
cells[i] = s
}
c.Writer.Write([]byte(strings.Join(cells, ",") + "\n"))
}
}
c.Writer.(http.Flusher).Flush()
rowCount := len(rows)
saveQueryHistory(0, int(instance.ID), dbName, queryStr, 0, rowCount)
} }
func saveQueryHistory(userID int, instanceID int, database string, sqlStr string, costMs int64, rowCount int) { func saveQueryHistory(userID int, instanceID int, database string, sqlStr string, costMs int64, rowCount int) {
+1 -1
View File
@@ -494,7 +494,7 @@ export default function QueriesPage() {
<span>{result.columns?.length || 0} 列</span> <span>{result.columns?.length || 0} 列</span>
</> </>
)} )}
{result.rows?.length > 0 && dbType !== 'redis' && ( {result.rows?.length > 0 && (
<button onClick={handleExportResult} <button onClick={handleExportResult}
className="dg-btn dg-btn-ghost dg-btn-sm"> className="dg-btn dg-btn-ghost dg-btn-sm">
<Download className="w-3 h-3" />导出 CSV <Download className="w-3 h-3" />导出 CSV